From b4bc4f382e3f194f9ed4da51c18e605d0fe52809 Mon Sep 17 00:00:00 2001 From: imccyu <276526105+imccyu@users.noreply.github.com> Date: Tue, 28 Jul 2026 22:45:35 +0800 Subject: [PATCH] refactor(session-projection): compact checkpoint row fields to ver/seq/val MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The persisted row (sessionId, key, stateVersion, observedSeq, state) becomes (sessionId, key, ver, seq, val) — the cache medium repeats these three names for every unit of every session, so the long forms dominated the JSON payload. ProjectionCheckpointRow and the checkpointRow zod spec rename together; the domain spec bumps to v3 (cache semantics: the old medium is discarded, not migrated). The unit-facing declaration keeps stateVersion — only the persisted/checkpoint row shape changes. --- ...ssion-projection-and-command-log.i18n.yaml | 4 +- ...7-27-session-projection-and-command-log.md | 4 +- ...7-session-projection-and-command-log.zh.md | 4 +- docs/cordis-catalog/services.md | 18 ++--- .../cordis/tool-cordis/src/api-catalog.ts | 10 +-- .../session-projection-cache/README.i18n.yaml | 4 +- .../session-projection-cache/README.md | 4 +- .../session-projection-cache/README.zh.md | 4 +- .../session-projection-cache/src/index.ts | 6 +- .../session-projection-cache/src/invariant.ts | 2 +- .../session-projection-cache/src/spec.ts | 23 +++---- .../tests/cache.spec.ts | 34 +++++----- .../session-projection/README.i18n.yaml | 4 +- .../session-projection/README.md | 4 +- .../session-projection/README.zh.md | 4 +- .../session-projection/src/index.ts | 65 +++++++++---------- .../session-projection/tests/registry.spec.ts | 48 +++++++------- 17 files changed, 121 insertions(+), 121 deletions(-) diff --git a/.agents/notes/proposed/architecture/2026-07-27-session-projection-and-command-log.i18n.yaml b/.agents/notes/proposed/architecture/2026-07-27-session-projection-and-command-log.i18n.yaml index 2476db5785..8f720e33b5 100644 --- a/.agents/notes/proposed/architecture/2026-07-27-session-projection-and-command-log.i18n.yaml +++ b/.agents/notes/proposed/architecture/2026-07-27-session-projection-and-command-log.i18n.yaml @@ -2,5 +2,5 @@ # side as of the last confirmed-consistent state. Both languages carry equal authority; # after editing either side, bring the other along and re-record with: # pnpm run verify-translation-pairing --write .agents/notes/proposed/architecture/2026-07-27-session-projection-and-command-log.md -2026-07-27-session-projection-and-command-log.md: 51cc60208ecafd55738c12f1887056c7b0427117 -2026-07-27-session-projection-and-command-log.zh.md: 71f6f6ea944c7c1bdd7e560ec8f0dc2528522fc1 +2026-07-27-session-projection-and-command-log.md: 6a073c956c27bbfc65cff2d4f44ca12023df0cd5 +2026-07-27-session-projection-and-command-log.zh.md: 500f07968db049e4a174ff3b7a075bfe095283db diff --git a/.agents/notes/proposed/architecture/2026-07-27-session-projection-and-command-log.md b/.agents/notes/proposed/architecture/2026-07-27-session-projection-and-command-log.md index 51cc60208e..6a073c956c 100644 --- a/.agents/notes/proposed/architecture/2026-07-27-session-projection-and-command-log.md +++ b/.agents/notes/proposed/architecture/2026-07-27-session-projection-and-command-log.md @@ -51,7 +51,7 @@ declare module 'cordis' { - Values are wire JSON payloads; the same map typed end to end (host unit, wire block, React hook) via `import type` — no second DTO table, no separate client-side "views" map. How a value is *rendered* is the slot system's business, never the projection layer's. - **The host is the only place a projection is computed.** The framework drives every registered unit forward eagerly: each committed session event passes through `apply`; a unit uninterested in an event returns the same state reference, and an unchanged reference (`Object.is`) produces no downstream work. Clients never fold domain events — they receive finished values (baseline block + push frame below). This removes the double-implementation trap (plan's two-event fold written once, on the host) and any client-side domain code. -- **State is always computed, never logged.** The log holds events only; the unit's state lives in the framework's per-session watermark cache (`{state, observedSeq}` per unit) and, in a later phase, in a **persisted projection cache** on the domain-KV storage seam: rows of `(sessionId, key, stateVersion, observedSeq, stateJson)`. A row is never wrong, only possibly stale — `observedSeq` says exactly how stale. The one read recipe, cold and live alike: take the cached state (or `init()`), forward-apply only the events past its watermark, `view` the result. Cold listings (every session's title across all workspaces) become an index read plus, at worst, a short tail replay; the session-persistence seam grows a read-from-seq primitive for that tail in the same later phase. Write policy: throttled (count/interval, configurable) plus two mandatory points — `turn/end` and detach (the live-to-cold moment). A crash between writes costs a longer tail replay, never a wrong value. +- **State is always computed, never logged.** The log holds events only; the unit's state lives in the framework's per-session watermark cache (`{state, observedSeq}` per unit) and, in a later phase, in a **persisted projection cache** on the domain-KV storage seam: rows of `(sessionId, key, ver, seq, val)` (`ver` = the unit's `stateVersion`, `seq` = the watermark, `val` = the state JSON). A row is never wrong, only possibly stale — its `seq` says exactly how stale. The one read recipe, cold and live alike: take the cached state (or `init()`), forward-apply only the events past its watermark, `view` the result. Cold listings (every session's title across all workspaces) become an index read plus, at worst, a short tail replay; the session-persistence seam grows a read-from-seq primitive for that tail in the same later phase. Write policy: throttled (count/interval, configurable) plus two mandatory points — `turn/end` and detach (the live-to-cold moment). A crash between writes costs a longer tail replay, never a wrong value. - A domain's input event set is its own choice: todos folds `todo/write` alone; plan folds `plan/mode` plus its own `/plan` `command/run` records (see the plan section); goal folds `goal/change` metadata; session title folds its title events (retiring the bespoke `session/title` frame and the client's title-snapshot map — the fourth hand-rolled projection this seam absorbs). - Registration is an effect (disposer with the fiber): an unloaded plugin's key disappears from subsequent responses and the client reads it as capability absence — HMR semantics for free. Duplicate keys throw. Domain plugins register under `ctx.inject(['sessionProjections'], …)` so headless assemblies without the registry stay unaffected. - The package owns `./invariant` (every served key has a live registration). @@ -133,7 +133,7 @@ Infrastructure first; the three in-flight PRs are left untouched and re-target a 2. **Client base**: the generic value store + `useProjection` seat; retire the per-domain cell machinery and, with title's unit registered, the `session/title` frame and title-snapshot map. Depends on 1 for the frame shape (fixtures feed synthetic frames meanwhile). 3. **Command channel**: the two events, executor logging, generic node + keyed slot, notice retirement, `{matched, commandId?}` admission. Parallel with 1. 4. **Domain re-targets** (after 1+2): todo (unit in `tool-todo`, drop the rider field), then plan (two-event unit, RPCs retired, toggle → `/plan`), then goal (`goal/change` unit, drop `goals.get`, move the six `Session` methods into the domain plugin's inject). -5. **Persisted projection cache** (later phase, after the domain-KV storage seam): the `(sessionId, key, stateVersion, observedSeq, state)` rows, throttled writes with turn/end + detach mandatory points, and the persistence read-from-seq primitive for cold tail replay. +5. **Persisted projection cache** (later phase, after the domain-KV storage seam): the `(sessionId, key, ver, seq, val)` rows, throttled writes with turn/end + detach mandatory points, and the persistence read-from-seq primitive for cold tail replay. ## Alternatives considered diff --git a/.agents/notes/proposed/architecture/2026-07-27-session-projection-and-command-log.zh.md b/.agents/notes/proposed/architecture/2026-07-27-session-projection-and-command-log.zh.md index 71f6f6ea94..500f07968d 100644 --- a/.agents/notes/proposed/architecture/2026-07-27-session-projection-and-command-log.zh.md +++ b/.agents/notes/proposed/architecture/2026-07-27-session-projection-and-command-log.zh.md @@ -51,7 +51,7 @@ declare module 'cordis' { - 值就是协议层的 JSON 载荷；同一张类型表经 `import type` 端到端贯通（host 侧单元、协议块、React 钩子）——没有第二张 DTO 表，也没有独立的客户端「views」表。值如何*渲染*是 slot 体系的事，永远不归投影层管。 - **host 是投影唯一的计算地点。** 框架正向驱动（eager drive）每个已注册的单元：每个已提交的会话事件都经过 `apply`；对某事件不感兴趣的单元返回同一个状态引用，而引用未变（`Object.is`）就不产生任何下游工作。客户端从不折叠领域事件——它们收到的是成品值（基线块 + 下文的推送帧）。这消除了双重实现陷阱（plan 的双事件折叠只在 host 写一遍），也消除了一切客户端侧领域代码。 -- **状态永远靠计算得出，绝不入日志。** 日志只存事件；单元的状态住在框架的按会话水位线缓存里（每单元一份 `{state, observedSeq}`），并在后续阶段进入 domain-KV 存储 seam 上的**持久投影缓存（persisted projection cache）**：形如 `(sessionId, key, stateVersion, observedSeq, stateJson)` 的行。一行永远不会是错的，至多是陈旧的——`observedSeq` 精确说明陈旧到哪。冷读与活读共用同一套读取配方：取缓存状态（或 `init()`），只对超出其水位线的事件做正向 `apply`，再对结果做 `view`。冷列表（跨全部 workspace 列出每个会话的标题）变成一次索引读，至多外加一小段尾部回放；session-persistence seam 在同一后续阶段为这段尾部补一个按 seq 起读的原语。写入策略：节流（次数/间隔，可配置）外加两个强制点——`turn/end` 与 detach（由活转冷的时刻）。两次写入之间崩溃的代价是尾部回放更长一些，绝不会是值出错。 +- **状态永远靠计算得出，绝不入日志。** 日志只存事件；单元的状态住在框架的按会话水位线缓存里（每单元一份 `{state, observedSeq}`），并在后续阶段进入 domain-KV 存储 seam 上的**持久投影缓存（persisted projection cache）**：形如 `(sessionId, key, ver, seq, val)` 的行（`ver` = 单元的 `stateVersion`，`seq` = 水位线，`val` = 状态 JSON）。一行永远不会是错的，至多是陈旧的——其 `seq` 精确说明陈旧到哪。冷读与活读共用同一套读取配方：取缓存状态（或 `init()`），只对超出其水位线的事件做正向 `apply`，再对结果做 `view`。冷列表（跨全部 workspace 列出每个会话的标题）变成一次索引读，至多外加一小段尾部回放；session-persistence seam 在同一后续阶段为这段尾部补一个按 seq 起读的原语。写入策略：节流（次数/间隔，可配置）外加两个强制点——`turn/end` 与 detach（由活转冷的时刻）。两次写入之间崩溃的代价是尾部回放更长一些，绝不会是值出错。 - 领域的输入事件集由领域自己选择：todos 只折叠 `todo/write`；plan 折叠 `plan/mode` 外加它自己的 `/plan` `command/run` 记录（见 plan 一节）；goal 折叠 `goal/change` 元数据；会话标题折叠其标题事件（顺带下线专设的 `session/title` 帧与客户端的标题快照表——这是该 seam 收编的第四个手工投影）。 - 注册是 effect（disposer 随 fiber 走）：插件卸载后其 key 从后续响应中消失，客户端将其读作能力缺失——HMR（热模块替换）语义随之自动成立。key 重复直接 throw。领域插件在 `ctx.inject(['sessionProjections'], …)` 下注册，因此不带注册表的 headless 组装完全不受影响。 - 该包拥有 `./invariant`（每个被服务的 key 都有一条存活的注册）。 @@ -133,7 +133,7 @@ host 侧命令执行器（`packages/ui/commands`）在调用处理器前追加 ` 2. **客户端基座**：通用值仓 + `useProjection` 席位；下线按领域的 cell 机制，并在标题单元注册后一并下线 `session/title` 帧与标题快照表。帧的形状依赖 1（在此之前 fixture（测试前置数据）喂合成帧）。 3. **命令通道**：两个事件、执行器落日志、通用节点 + keyed slot、通知通道下线、`{matched, commandId?}` 准入。与 1 并行。 4. **领域重新对接**（在 1+2 之后）：先 todo（单元进 `tool-todo`，删掉搭载字段），再 plan（双事件单元、RPC 下线、开关改发 `/plan`），最后 goal（`goal/change` 单元，删掉 `goals.get`，把六个 `Session` 方法移入领域插件的 inject）。 -5. **持久投影缓存**（后续阶段，待 domain-KV 存储 seam 就绪后）：`(sessionId, key, stateVersion, observedSeq, state)` 行、带 turn/end 与 detach 强制点的节流写入，以及持久化侧供冷尾部回放用的按 seq 起读原语。 +5. **持久投影缓存**（后续阶段，待 domain-KV 存储 seam 就绪后）：`(sessionId, key, ver, seq, val)` 行、带 turn/end 与 detach 强制点的节流写入，以及持久化侧供冷尾部回放用的按 seq 起读原语。 ## Alternatives considered diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index e6895f6a8d..eef530b0df 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -1191,9 +1191,9 @@ snapshot(session: Session): ProjectionSnapshot * State-level checkpoint of every registered unit for one session, read * from the watermark cache (missing cells fold lazily over the in-memory * log). This is the write side of the persisted projection cache: the - * returned rows are the `(key → {stateVersion, observedSeq, state})` part - * of the durable `(sessionId, key, stateVersion, observedSeq, state)` - * rows. Every `state` is a DETACHED structured clone — never the live + * returned rows are the `(key → {ver, seq, val})` part of the durable + * `(sessionId, key, ver, seq, val)` + * rows. Every `val` is a DETACHED structured clone — never the live * cell reference: the watermark cache is this registry's authoritative * mutable state, and a caller reaching the live reference could corrupt * every subsequent snapshot and frame through it (plain JSON by the unit @@ -1206,7 +1206,7 @@ checkpoint(session: Session): ProjectionCheckpoint /** * The stored seq a {@link restore} tail read over `checkpoint` must start * at: one event BELOW the lowest usable watermark (a row is usable when - * its `stateVersion` matches the live unit; an absent or mismatched row + * its `ver` matches the live unit's `stateVersion`; an absent or mismatched row * pulls the floor to `0` — that key must refold the full log). The * one-below anchor is load-bearing: the tail then proves how far the * stored log still extends, so {@link restore} can detect a log that @@ -1223,7 +1223,7 @@ restoreFloor(checkpoint: ProjectionCheckpoint): number | undefined /** * View a checkpoint's rows without any log read: for every registered - * unit whose row's `stateVersion` matches, serve the schema-validated + * unit whose row's `ver` matches, serve the schema-validated * `view` of the stored state; mismatched or absent rows leave their key * absent (a cold or listing consumer treats it as not-yet-available and a * fuller read path refolds it). The zero-I/O rung of the read ladder — @@ -1241,9 +1241,9 @@ viewCheckpoint(checkpoint: ProjectionCheckpoint): Partial * `readFrom(id, restoreFloor(checkpoint))` and that same floor as * `baseSeq`; the floor's one-below anchor makes the supplied end honest, * so a shrunk log is detected here. A row is usable iff its - * `stateVersion` matches the live unit, it does not predate `baseSeq` - * (`observedSeq >= baseSeq - 1`), and it does not claim events past the - * supplied end (`observedSeq <= endSeq`); an unusable row is discarded + * `ver` matches the live unit's `stateVersion`, it does not predate `baseSeq` + * (`seq >= baseSeq - 1`), and it does not claim events past the + * supplied end (`seq <= endSeq`); an unusable row is discarded * and its key refolds from `init` — which is only sound over the full * log, so a discarded row with `baseSeq > 0` throws (the caller re-reads * from seq 0, e.g. after a crash-repair truncation shrank the log below @@ -1260,7 +1260,7 @@ restore(checkpoint: ProjectionCheckpoint, events: readonly SessionEvent[], baseS Types: [Session](../core-data-structures/session.md) · [SessionEvent](../core-data-structures/core.md) -Source: [`packages/session-projection/session-projection/src/index.ts:157`](../../packages/session-projection/session-projection/src/index.ts) +Source: [`packages/session-projection/session-projection/src/index.ts:156`](../../packages/session-projection/session-projection/src/index.ts) ## `ctx.sessionQuery` — `SessionQueryService` (abstract seam) diff --git a/packages/cordis/tool-cordis/src/api-catalog.ts b/packages/cordis/tool-cordis/src/api-catalog.ts index fb435e7ca2..c1011c2326 100644 --- a/packages/cordis/tool-cordis/src/api-catalog.ts +++ b/packages/cordis/tool-cordis/src/api-catalog.ts @@ -574,19 +574,19 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [ }, { signature: 'checkpoint(session: Session): ProjectionCheckpoint', - jsDoc: '/**\n * State-level checkpoint of every registered unit for one session, read\n * from the watermark cache (missing cells fold lazily over the in-memory\n * log). This is the write side of the persisted projection cache: the\n * returned rows are the `(key → {stateVersion, observedSeq, state})` part\n * of the durable `(sessionId, key, stateVersion, observedSeq, state)`\n * rows. Every `state` is a DETACHED structured clone — never the live\n * cell reference: the watermark cache is this registry\'s authoritative\n * mutable state, and a caller reaching the live reference could corrupt\n * every subsequent snapshot and frame through it (plain JSON by the unit\n * contract, so the clone is total).\n * @param session - the session whose unit states are checkpointed.\n * @returns one row per registered key; empty when no unit is registered.\n */', + jsDoc: '/**\n * State-level checkpoint of every registered unit for one session, read\n * from the watermark cache (missing cells fold lazily over the in-memory\n * log). This is the write side of the persisted projection cache: the\n * returned rows are the `(key → {ver, seq, val})` part of the durable\n * `(sessionId, key, ver, seq, val)`\n * rows. Every `val` is a DETACHED structured clone — never the live\n * cell reference: the watermark cache is this registry\'s authoritative\n * mutable state, and a caller reaching the live reference could corrupt\n * every subsequent snapshot and frame through it (plain JSON by the unit\n * contract, so the clone is total).\n * @param session - the session whose unit states are checkpointed.\n * @returns one row per registered key; empty when no unit is registered.\n */', }, { signature: 'restoreFloor(checkpoint: ProjectionCheckpoint): number | undefined', - jsDoc: '/**\n * The stored seq a {@link restore} tail read over `checkpoint` must start\n * at: one event BELOW the lowest usable watermark (a row is usable when\n * its `stateVersion` matches the live unit; an absent or mismatched row\n * pulls the floor to `0` — that key must refold the full log). The\n * one-below anchor is load-bearing: the tail then proves how far the\n * stored log still extends, so {@link restore} can detect a log that\n * shrank below a row\'s watermark (crash-repair truncation) instead of\n * serving the stale row as current — an empty tail read from the anchor\n * yields an end below every watermark and the restore rejects for a full\n * re-read.\n * @param checkpoint - persisted rows for one session (possibly stale or empty).\n * @returns the seq to hand the persistence `readFrom`, or `undefined`\n * when no unit is registered (no read needed — {@link restore} would\n * serve empty values regardless).\n */', + jsDoc: '/**\n * The stored seq a {@link restore} tail read over `checkpoint` must start\n * at: one event BELOW the lowest usable watermark (a row is usable when\n * its `ver` matches the live unit\'s `stateVersion`; an absent or mismatched row\n * pulls the floor to `0` — that key must refold the full log). The\n * one-below anchor is load-bearing: the tail then proves how far the\n * stored log still extends, so {@link restore} can detect a log that\n * shrank below a row\'s watermark (crash-repair truncation) instead of\n * serving the stale row as current — an empty tail read from the anchor\n * yields an end below every watermark and the restore rejects for a full\n * re-read.\n * @param checkpoint - persisted rows for one session (possibly stale or empty).\n * @returns the seq to hand the persistence `readFrom`, or `undefined`\n * when no unit is registered (no read needed — {@link restore} would\n * serve empty values regardless).\n */', }, { signature: 'viewCheckpoint(checkpoint: ProjectionCheckpoint): Partial', - jsDoc: '/**\n * View a checkpoint\'s rows without any log read: for every registered\n * unit whose row\'s `stateVersion` matches, serve the schema-validated\n * `view` of the stored state; mismatched or absent rows leave their key\n * absent (a cold or listing consumer treats it as not-yet-available and a\n * fuller read path refolds it). The zero-I/O rung of the read ladder —\n * values are as stale as their rows, never wrong.\n * @param checkpoint - persisted rows for one session (possibly stale or empty).\n * @returns whole values per key with a usable row; empty when none.\n */', + jsDoc: '/**\n * View a checkpoint\'s rows without any log read: for every registered\n * unit whose row\'s `ver` matches, serve the schema-validated\n * `view` of the stored state; mismatched or absent rows leave their key\n * absent (a cold or listing consumer treats it as not-yet-available and a\n * fuller read path refolds it). The zero-I/O rung of the read ladder —\n * values are as stale as their rows, never wrong.\n * @param checkpoint - persisted rows for one session (possibly stale or empty).\n * @returns whole values per key with a usable row; empty when none.\n */', }, { signature: 'restore(checkpoint: ProjectionCheckpoint, events: readonly SessionEvent[], baseSeq: number): { snapshot: ProjectionSnapshot; checkpoint: ProjectionCheckpoint }', - jsDoc: '/**\n * Cold read: fold every registered unit over a stored log suffix, seeding\n * each from its checkpoint row when usable — the one read recipe (cached\n * state + forward tail replay + `view`) applied without a live `Session`.\n * Call with the events returned by a persistence\n * `readFrom(id, restoreFloor(checkpoint))` and that same floor as\n * `baseSeq`; the floor\'s one-below anchor makes the supplied end honest,\n * so a shrunk log is detected here. A row is usable iff its\n * `stateVersion` matches the live unit, it does not predate `baseSeq`\n * (`observedSeq >= baseSeq - 1`), and it does not claim events past the\n * supplied end (`observedSeq <= endSeq`); an unusable row is discarded\n * and its key refolds from `init` — which is only sound over the full\n * log, so a discarded row with `baseSeq > 0` throws (the caller re-reads\n * from seq 0, e.g. after a crash-repair truncation shrank the log below\n * a row\'s watermark).\n * @param checkpoint - persisted rows for one session (possibly stale or empty).\n * @param events - the stored events with `seq >= baseSeq`, in seq order.\n * @param baseSeq - the seq `events` starts at (its first event\'s seq when non-empty).\n * @returns the snapshot cut at the supplied log end (`asOfSeq` is the last\n * supplied event\'s seq, `baseSeq - 1` for an empty tail) plus the\n * refreshed checkpoint rows at that cut, ready for a durable write-back.\n */', + jsDoc: '/**\n * Cold read: fold every registered unit over a stored log suffix, seeding\n * each from its checkpoint row when usable — the one read recipe (cached\n * state + forward tail replay + `view`) applied without a live `Session`.\n * Call with the events returned by a persistence\n * `readFrom(id, restoreFloor(checkpoint))` and that same floor as\n * `baseSeq`; the floor\'s one-below anchor makes the supplied end honest,\n * so a shrunk log is detected here. A row is usable iff its\n * `ver` matches the live unit\'s `stateVersion`, it does not predate `baseSeq`\n * (`seq >= baseSeq - 1`), and it does not claim events past the\n * supplied end (`seq <= endSeq`); an unusable row is discarded\n * and its key refolds from `init` — which is only sound over the full\n * log, so a discarded row with `baseSeq > 0` throws (the caller re-reads\n * from seq 0, e.g. after a crash-repair truncation shrank the log below\n * a row\'s watermark).\n * @param checkpoint - persisted rows for one session (possibly stale or empty).\n * @param events - the stored events with `seq >= baseSeq`, in seq order.\n * @param baseSeq - the seq `events` starts at (its first event\'s seq when non-empty).\n * @returns the snapshot cut at the supplied log end (`asOfSeq` is the last\n * supplied event\'s seq, `baseSeq - 1` for an empty tail) plus the\n * refreshed checkpoint rows at that cut, ready for a durable write-back.\n */', }, ], }, @@ -1899,7 +1899,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [ }, { name: 'ProjectionCheckpointRow', - declaration: 'export interface ProjectionCheckpointRow {\n stateVersion: number;\n observedSeq: number;\n state: unknown;\n}', + declaration: 'export interface ProjectionCheckpointRow {\n ver: number;\n seq: number;\n val: unknown;\n}', }, { name: 'ProjectionDefinition', diff --git a/packages/session-projection/session-projection-cache/README.i18n.yaml b/packages/session-projection/session-projection-cache/README.i18n.yaml index 2214de288d..c1bdd6c5b8 100644 --- a/packages/session-projection/session-projection-cache/README.i18n.yaml +++ b/packages/session-projection/session-projection-cache/README.i18n.yaml @@ -2,5 +2,5 @@ # side as of the last confirmed-consistent state. Both languages carry equal authority; # after editing either side, bring the other along and re-record with: # pnpm run verify-translation-pairing --write packages/session-projection/session-projection-cache/README.md -README.md: a10bb858159e9581c815d2532a893208c45e788a -README.zh.md: 191efbc2c70875b20f3d5d2e873c87a3bd001eff +README.md: 5d4ad07fab6648acdb40c6aa86d32cc78b4c016e +README.zh.md: ab4076df28cfe2b5a8b41d609967039dcacb7ef4 diff --git a/packages/session-projection/session-projection-cache/README.md b/packages/session-projection/session-projection-cache/README.md index a10bb85815..5d4ad07fab 100644 --- a/packages/session-projection/session-projection-cache/README.md +++ b/packages/session-projection/session-projection-cache/README.md @@ -4,10 +4,10 @@ English | [中文](README.zh.md) The persisted projection cache (`ctx.sessionProjectionCache`): durable checkpoints of every registered projection unit's state, one record per session on the domain data form (`session_projcache` domain — the shipped json backend lands it beside `workspace.json` under the configured storage root). Design authority: the [session-projection RFC](../../../.agents/notes/proposed/architecture/2026-07-27-session-projection-and-command-log.md) (persisted projection cache section). -A stored row `(key → {stateVersion, observedSeq, state})` is a fold shortcut, never an authority: possibly stale (`observedSeq` says exactly how stale) but never wrong. Consequences the implementation commits to: +A stored row `(key → {ver, seq, val})` is a fold shortcut, never an authority: possibly stale (`seq` says exactly how stale) but never wrong. Consequences the implementation commits to: - **Every background write is fail-soft.** A failed durable write logs a warning and keeps the cache stale; the next write or cold read self-heals. A crash between writes costs a longer tail replay, never a wrong value. -- **`stateVersion` mismatch discards, never migrates.** A unit bump invalidates its rows at read time; the key refolds from the log. +- **A `ver` mismatch against the live unit's `stateVersion` discards, never migrates.** A unit bump invalidates its rows at read time; the key refolds from the log. - **Whole-record writes.** Each write replaces the session's full checkpoint (the registry cut is always complete), snapshotted through the lossless-JSON boundary — a unit state violating the plain-JSON contract fails loud. - **Records are bound to a log lifecycle, not just an id.** Each record stores the header identity (`createdAt`, `cwd`) it was folded from; every read validates it (the live or stored header is the witness) before accepting a row, so a deleted-then-recreated id or a persistence store swapped under a surviving cache discards the unrelated record instead of seeding phantom values. - **The log leads, the cache follows.** A live checkpoint flushes the session's buffered events durably BEFORE the cache row lands, so a crash can leave the cache behind the log (a longer tail replay) but never ahead of it. diff --git a/packages/session-projection/session-projection-cache/README.zh.md b/packages/session-projection/session-projection-cache/README.zh.md index 191efbc2c7..ab4076df28 100644 --- a/packages/session-projection/session-projection-cache/README.zh.md +++ b/packages/session-projection/session-projection-cache/README.zh.md @@ -4,10 +4,10 @@ 持久投影缓存（`ctx.sessionProjectionCache`）：把每个已注册投影单元的状态持久化为检查点（checkpoint），基于域数据形态（domain data form）每会话一条记录（`session_projcache` 域——出厂 json 后端将其落在配置的存储根目录下、`workspace.json` 旁边）。设计权威：[session-projection RFC](../../../.agents/notes/proposed/architecture/2026-07-27-session-projection-and-command-log.md)（persisted projection cache 一节）。 -一条存储行 `(key → {stateVersion, observedSeq, state})` 是折叠捷径，绝不是权威：可能陈旧（`observedSeq` 精确说明陈旧到哪），但绝不会错。实现据此承诺： +一条存储行 `(key → {ver, seq, val})` 是折叠捷径，绝不是权威：可能陈旧（`seq` 精确说明陈旧到哪），但绝不会错。实现据此承诺： - **每次后台写入都 fail-soft。** 持久写失败只记一条警告并保持缓存陈旧；下一次写入或冷读自愈。两次写之间崩溃的代价是更长的尾部重放，绝不是错误的值。 -- **`stateVersion` 不匹配即丢弃，绝不迁移。** 单元递增版本会在读取时使其行失效；该 key 从日志重新折叠。 +- **`ver` 与活单元 `stateVersion` 不匹配即丢弃，绝不迁移。** 单元递增版本会在读取时使其行失效；该 key 从日志重新折叠。 - **整记录写入。** 每次写入替换该会话的完整检查点（注册表切面始终是完整的），并经无损 JSON 边界快照——违反纯 JSON 契约的单元状态会大声失败。 - **记录绑定到日志生命周期，而不只是 id。** 每条记录存储其折叠来源的 header 身份（`createdAt`、`cwd`）；每次读取先以活 header 或存储 header 为证验证它,再接受任何行——被删后重建的 id、或缓存幸存而持久化存储被换掉时,无关记录被整体丢弃,绝不播种幻影值。 - **日志领先，缓存跟随。** 活会话检查点先把缓冲事件持久 flush,缓存行才落地,因此崩溃只会让缓存落后于日志（更长的尾部重放）,绝不领先于它。 diff --git a/packages/session-projection/session-projection-cache/src/index.ts b/packages/session-projection/session-projection-cache/src/index.ts index c598bd1312..03b3ed26ee 100644 --- a/packages/session-projection/session-projection-cache/src/index.ts +++ b/packages/session-projection/session-projection-cache/src/index.ts @@ -3,10 +3,10 @@ * checkpoints of every registered projection unit's state, one record per * session on the domain data form (`session_projcache` domain — the shipped * json backend lands it beside `workspace.json`). The cache is a fold - * shortcut, never an authority: a row is possibly stale (its `observedSeq` + * shortcut, never an authority: a row is possibly stale (its `seq` * says how stale) but never wrong, so every write path is fail-soft (a lost * write costs a longer tail replay on the next cold read) and a - * `stateVersion` mismatch discards the row instead of migrating it. Design + * `ver` mismatch discards the row instead of migrating it. Design * authority: the session-projection RFC * (.agents/notes/proposed/architecture/2026-07-27-session-projection-and-command-log.md). * @module @deepseek-ai/dsh-session-projection-cache @@ -125,7 +125,7 @@ export class SessionProjectionCache extends Service { // The block carries ONE cut: the lowest served watermark is the seq every // value is at least current as of (under-claiming is safe under // higher-seq-wins; over-claiming would let a stale value outrank pushes). - const asOfSeq = Math.min(...keys.map(key => (record.rows[key] as { observedSeq: number }).observedSeq)) + const asOfSeq = Math.min(...keys.map(key => (record.rows[key] as { seq: number }).seq)) return { asOfSeq, values } } diff --git a/packages/session-projection/session-projection-cache/src/invariant.ts b/packages/session-projection/session-projection-cache/src/invariant.ts index a5a5515f22..8a119044d8 100644 --- a/packages/session-projection/session-projection-cache/src/invariant.ts +++ b/packages/session-projection/session-projection-cache/src/invariant.ts @@ -16,7 +16,7 @@ export const inject = ['invariants'] /** * No runtime invariant: the cache's correctness relation (a stored row equals - * the registry fold at its `observedSeq`) is only checkable by re-running the + * the registry fold at its `seq` watermark) is only checkable by re-running the * fold over the persisted log — duplicating the implementation rather than * detecting drift — and its staleness is by design (fail-soft writes). The * durable boundary is already schema-validated by the storage-domain layer diff --git a/packages/session-projection/session-projection-cache/src/spec.ts b/packages/session-projection/session-projection-cache/src/spec.ts index 72796b6c4b..5e41763f80 100644 --- a/packages/session-projection/session-projection-cache/src/spec.ts +++ b/packages/session-projection/session-projection-cache/src/spec.ts @@ -1,7 +1,7 @@ /** * The session-projcache domain declaration: one `sessions` table keyed by * {@link SessionId}, each record the full projection checkpoint for one - * session (`key → {stateVersion, observedSeq, state}` rows). The spec object + * session (`key → {ver, seq, val}` rows). The spec object * is the single source of the domain's identity, version, and record schema; * the storage-domain routing decides the medium (the shipped composition's * json backend lands it at `/session_projcache.json`, beside @@ -14,17 +14,17 @@ import { SessionId } from '@deepseek-ai/dsh-session' import { defineDomain, domainTable } from '@deepseek-ai/dsh-storage-domain' /** - * One persisted checkpoint row (the RFC's `(sessionId, key, stateVersion, - * observedSeq, state)` minus the two record keys). `state` is the unit's - * internal state — plain JSON by the unit contract; `z.json()` enforces that - * at the durable boundary. A row is never wrong, only possibly stale: - * `observedSeq` says exactly how stale, and a `stateVersion` mismatch + * One persisted checkpoint row (the RFC's `(sessionId, key, ver, seq, val)` + * minus the two record keys). `val` is the unit's internal state — plain + * JSON by the unit contract; `z.json()` enforces that at the durable + * boundary. A row is never wrong, only possibly stale: `seq` says exactly + * how stale, and a `ver` mismatch against the live unit's `stateVersion` * discards it at read time (never a migration). */ export const checkpointRow = z.object({ - stateVersion: z.number().int().nonnegative(), - observedSeq: z.number().int().gte(-1), - state: z.json(), + ver: z.number().int().nonnegative(), + seq: z.number().int().gte(-1), + val: z.json(), }) /** @@ -61,10 +61,11 @@ export type CheckpointRecord = z.infer /** * The session-projcache domain spec. Version bumps discard the whole medium * (cache semantics: a stale or unreadable cache costs a longer tail replay, - * never a wrong value). v2 added the record's log-identity binding. + * never a wrong value). v2 added the record's log-identity binding; v3 + * renamed the row fields to `ver`/`seq`/`val`. */ export const projectionCacheDomainSpec = defineDomain({ name: 'session_projcache', - version: 2, + version: 3, tables: { sessions: domainTable(checkpointRecord) }, }) diff --git a/packages/session-projection/session-projection-cache/tests/cache.spec.ts b/packages/session-projection/session-projection-cache/tests/cache.spec.ts index c7c7b8502f..8cf9345772 100644 --- a/packages/session-projection/session-projection-cache/tests/cache.spec.ts +++ b/packages/session-projection/session-projection-cache/tests/cache.spec.ts @@ -100,7 +100,7 @@ function storedRecord(pool: MemoryMediaPool, id: Session['id']) { return pool.media.get('session_projcache')?.tables.get('sessions')?.get(String(id)) as { identity: { createdAt: number; cwd?: string } - rows: Record + rows: Record } | undefined } @@ -126,7 +126,7 @@ describe('SessionProjectionCache write policy', () => { const end = endTurn(session) await settle() const rows = storedRows(pool, session.id) - expect(rows?.['cache-test/marks']).toEqual({ stateVersion: 1, observedSeq: end.seq, state: { marks: ['a'] } }) + expect(rows?.['cache-test/marks']).toEqual({ ver: 1, seq: end.seq, val: { marks: ['a'] } }) }) it('writes at session disposal (detach, the live-to-cold moment)', async () => { @@ -140,7 +140,7 @@ describe('SessionProjectionCache write policy', () => { mark(session, ['live']) await owner.dispose() await settle() - expect(storedRows(pool, session.id)?.['cache-test/marks']?.state).toEqual({ marks: ['live'] }) + expect(storedRows(pool, session.id)?.['cache-test/marks']?.val).toEqual({ marks: ['live'] }) }) it('flushes when the in-turn event count reaches the configured threshold', async () => { @@ -152,7 +152,7 @@ describe('SessionProjectionCache write policy', () => { expect(storedRows(pool, session.id)).toBeUndefined() mark(session, ['3']) await settle() - expect(storedRows(pool, session.id)?.['cache-test/marks']?.state).toEqual({ marks: ['3'] }) + expect(storedRows(pool, session.id)?.['cache-test/marks']?.val).toEqual({ marks: ['3'] }) }) it('flushes on the configured interval when the count threshold is not reached', async () => { @@ -164,7 +164,7 @@ describe('SessionProjectionCache write policy', () => { expect(storedRows(pool, session.id)).toBeUndefined() await vi.advanceTimersByTimeAsync(1) await vi.advanceTimersByTimeAsync(0) - expect(storedRows(pool, session.id)?.['cache-test/marks']?.state).toEqual({ marks: ['slow'] }) + expect(storedRows(pool, session.id)?.['cache-test/marks']?.val).toEqual({ marks: ['slow'] }) }) it('write() on a never-dirty session checkpoints directly and rejects a non-JSON unit state', async () => { @@ -172,7 +172,7 @@ describe('SessionProjectionCache write policy', () => { // Never dirtied: no events — write() still lands the init-derived cut. const clean = ctx.sessions.create(SessionId('clean-write')) await ctx.sessionProjectionCache.write(clean) - expect(storedRows(pool, clean.id)?.['cache-test/marks']).toEqual({ stateVersion: 1, observedSeq: -1, state: null }) + expect(storedRows(pool, clean.id)?.['cache-test/marks']).toEqual({ ver: 1, seq: -1, val: null }) // A unit whose state violates the plain-JSON contract fails the write loud. ctx.sessionProjections.register({ key: 'cache-test/marks2' as never, @@ -214,7 +214,7 @@ describe('SessionProjectionCache write policy', () => { mark(session, ['y']) endTurn(session) await settle() - expect(storedRows(pool, session.id)?.['cache-test/marks']?.state).toEqual({ marks: ['y'] }) + expect(storedRows(pool, session.id)?.['cache-test/marks']?.val).toEqual({ marks: ['y'] }) }) }) @@ -234,10 +234,10 @@ describe('SessionProjectionCache cold read', () => { function seedRow( pool: MemoryMediaPool, id: string, - row: { stateVersion: number; observedSeq: number; state: unknown }, + row: { ver: number; seq: number; val: unknown }, identity: { createdAt: number; cwd?: string } = { createdAt: 0 }, ): void { - pool.versions.set('session_projcache', 2) + pool.versions.set('session_projcache', 3) pool.media.set('session_projcache', { tables: new Map([['sessions', new Map([[id, { identity, rows: { 'cache-test/marks': row } }]])]]), global: null, @@ -248,7 +248,7 @@ describe('SessionProjectionCache cold read', () => { const pool = new MemoryMediaPool() const logs = new Map([['cold', storedLog([['a'], ['a', 'b']])]]) // A warm-era checkpoint at watermark 1 (only ['a'] folded). - seedRow(pool, 'cold', { stateVersion: 1, observedSeq: 1, state: { marks: ['a'] } }) + seedRow(pool, 'cold', { ver: 1, seq: 1, val: { marks: ['a'] } }) const { cache, persistence, pool: samePool } = await harness({ pool, logs }) const id = SessionId('cold') const snapshot = await cache.coldSnapshot(id) @@ -258,13 +258,13 @@ describe('SessionProjectionCache cold read', () => { expect(persistence.readFrom).toHaveBeenCalledWith(id, 1, undefined) // Write-back: the stored row advanced to the served cut. expect(storedRows(samePool, id)?.['cache-test/marks']) - .toEqual({ stateVersion: 1, observedSeq: 3, state: { marks: ['a', 'b'] } }) + .toEqual({ ver: 1, seq: 3, val: { marks: ['a', 'b'] } }) }) it('discards a version-mismatched row and refolds the full log', async () => { const pool = new MemoryMediaPool() const logs = new Map([['bumped', storedLog([['a']])]]) - seedRow(pool, 'bumped', { stateVersion: 1, observedSeq: 2, state: { marks: ['stale'] } }) + seedRow(pool, 'bumped', { ver: 1, seq: 2, val: { marks: ['stale'] } }) const { cache, persistence } = await harness({ pool, logs, stateVersion: 2 }) const snapshot = await cache.coldSnapshot(SessionId('bumped')) expect(snapshot.values['cache-test/marks']).toEqual({ marks: ['a'] }) @@ -276,7 +276,7 @@ describe('SessionProjectionCache cold read', () => { it('detects a log shrunk below the row watermark and degrades to one full re-read', async () => { const pool = new MemoryMediaPool() const logs = new Map([['shrunk', storedLog([['a']])]]) // seqs 0..2 - seedRow(pool, 'shrunk', { stateVersion: 1, observedSeq: 9, state: { marks: ['ghost'] } }) + seedRow(pool, 'shrunk', { ver: 1, seq: 9, val: { marks: ['ghost'] } }) const { cache, persistence } = await harness({ pool, logs }) const snapshot = await cache.coldSnapshot(SessionId('shrunk')) expect(snapshot.values['cache-test/marks']).toEqual({ marks: ['a'] }) @@ -307,7 +307,7 @@ describe('SessionProjectionCache cold read', () => { const logs = new Map([['reborn', storedLog([['real']])]]) // stored header stamps createdAt 0 // A checkpoint from a PRIOR lifecycle of the same id (different createdAt): // its rows pass every watermark check, but the identity does not match. - seedRow(pool, 'reborn', { stateVersion: 1, observedSeq: 2, state: { marks: ['phantom'] } }, { createdAt: 999 }) + seedRow(pool, 'reborn', { ver: 1, seq: 2, val: { marks: ['phantom'] } }, { createdAt: 999 }) const { cache, pool: samePool } = await harness({ pool, logs }) const snapshot = await cache.coldSnapshot(SessionId('reborn')) expect(snapshot.values['cache-test/marks']).toEqual({ marks: ['real'] }) @@ -317,14 +317,14 @@ describe('SessionProjectionCache cold read', () => { it('cachedSnapshot returns undefined when every stored row is version-mismatched', async () => { const pool = new MemoryMediaPool() - seedRow(pool, 'all-stale', { stateVersion: 99, observedSeq: 4, state: { marks: ['old'] } }) + seedRow(pool, 'all-stale', { ver: 99, seq: 4, val: { marks: ['old'] } }) const { cache } = await harness({ pool }) expect(cache.cachedSnapshot(headerOf(SessionId('all-stale')))).toBeUndefined() }) it('binds identity on cwd too: a matching cwd serves, a moved session does not', async () => { const pool = new MemoryMediaPool() - seedRow(pool, 'homed', { stateVersion: 1, observedSeq: 2, state: { marks: ['w'] } }, { createdAt: 0, cwd: '/work' }) + seedRow(pool, 'homed', { ver: 1, seq: 2, val: { marks: ['w'] } }, { createdAt: 0, cwd: '/work' }) const { cache } = await harness({ pool }) const id = SessionId('homed') expect(cache.cachedSnapshot(headerOf(id, 0, '/work'))?.values['cache-test/marks']).toEqual({ marks: ['w'] }) @@ -352,7 +352,7 @@ describe('SessionProjectionCache cold read', () => { it('cachedSnapshot serves identity-matching rows with the cut watermark and refuses unrelated ones', async () => { const pool = new MemoryMediaPool() - seedRow(pool, 'listed', { stateVersion: 1, observedSeq: 4, state: { marks: ['t'] } }) + seedRow(pool, 'listed', { ver: 1, seq: 4, val: { marks: ['t'] } }) const { cache } = await harness({ pool }) const id = SessionId('listed') // Matching header: values plus the watermark the client seeds under. diff --git a/packages/session-projection/session-projection/README.i18n.yaml b/packages/session-projection/session-projection/README.i18n.yaml index 7a54d3214f..a15bee5bac 100644 --- a/packages/session-projection/session-projection/README.i18n.yaml +++ b/packages/session-projection/session-projection/README.i18n.yaml @@ -2,5 +2,5 @@ # side as of the last confirmed-consistent state. Both languages carry equal authority; # after editing either side, bring the other along and re-record with: # pnpm run verify-translation-pairing --write packages/session-projection/session-projection/README.md -README.md: 2e026aab55933c96ba961481f9597bc18cbbe910 -README.zh.md: a3e0b0f46466d19321b0950dc41d06473a54a1ce +README.md: f4898b8e567fa5998c18111c5f4e27a8a350a42e +README.zh.md: 385862868df495a5c857d91c32f6503c3ef72025 diff --git a/packages/session-projection/session-projection/README.md b/packages/session-projection/session-projection/README.md index 2e026aab55..f4898b8e56 100644 --- a/packages/session-projection/session-projection/README.md +++ b/packages/session-projection/session-projection/README.md @@ -23,7 +23,7 @@ Session-projection seam. It owns `ctx.sessionProjections`, the registry that DRI - **Same-reference means no work.** `apply` MUST return the same state reference for events that do not concern the unit; the drive gates the change feed on `Object.is`, so non-matching events cost one call and nothing downstream. - **Whole-value event rule (load-bearing).** A state-carrying log event MUST carry the complete post-change state, never a bare delta — it keeps every transition trivially cheap and every served value self-describing (last-wins for consumers). - **Synchronous unit discipline.** `init`/`apply`/`view` MUST be synchronous; 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, which fails the boundary `schema.parse` loudly. -- **State is plain JSON, `stateVersion` is its invalidation anchor.** The persisted projection cache (a later phase) stores `(sessionId, key, stateVersion, observedSeq, stateJson)` rows; bump `stateVersion` whenever the state shape or the fold semantics change so stale rows are discarded instead of forward-applied into garbage. +- **State is plain JSON, `stateVersion` is its invalidation anchor.** The persisted projection cache stores `(sessionId, key, ver, seq, val)` rows; bump `stateVersion` whenever the state shape or the fold semantics change so stale rows are discarded instead of forward-applied into garbage. - **No wire vocabulary here.** The registry exposes only the change feed and the snapshot read face; carriers (api-proxy) mint their own frames (`session/projection`) and blocks from them. - **Optional seam.** Domain plugins register under `ctx.inject(['sessionProjections'], …)` so headless assemblies without the registry stay unaffected; carriers use `ctx.get('sessionProjections')` and omit their block/frames entirely when the registry is absent. @@ -43,5 +43,5 @@ None; projections never assemble or send provider requests. - **Every tail page carries every registered key** — there is no per-key opt-out or lazy-key request shape yet; acceptable while values are UI-scale whole states (a todo list, a goal snapshot), revisit if a domain's value grows large. - **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, addable without contract change. -- **The persisted projection cache is a later phase** — cells live in memory only; a restart rebuilds by folding the in-memory log on first touch. The `stateVersion` field is the forward-declared invalidation anchor for that phase. +- **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** — the boundary `schema.parse` rejects a Promise-returning `view`, but an `apply` that blocks or reads torn non-session state is a review concern; the invariant companion documents why no runtime check exists. diff --git a/packages/session-projection/session-projection/README.zh.md b/packages/session-projection/session-projection/README.zh.md index a3e0b0f464..385862868d 100644 --- a/packages/session-projection/session-projection/README.zh.md +++ b/packages/session-projection/session-projection/README.zh.md @@ -23,7 +23,7 @@ - **同引用即无工作。** 对与单元无关的事件，`apply` 必须返回同一个状态引用；驱动以 `Object.is` 把守变更流，因此不匹配的事件只花一次调用，不产生任何下游工作。 - **全量值事件规则（承重）。** 携带状态的日志事件必须携带变更后的完整状态，绝不携带裸增量——这让每次状态转移始终足够廉价，也让每个被供给的值自描述（对消费方即 last-wins）。 - **单元的同步纪律。** `init`/`apply`/`view` 必须是同步的；载体在切出页面切片的同一 tick 内读取 `snapshot()`，`asOfSeq` 之所以是一个一致切面正系于此。误写成异步的 `view` 会返回 Promise，让边界的 `schema.parse` 当场大声失败。 -- **状态是纯 JSON，`stateVersion` 是其失效锚点。** 持久投影缓存（persisted projection cache，后续阶段）存储 `(sessionId, key, stateVersion, observedSeq, stateJson)` 行；状态形状或折叠语义一旦变化就递增 `stateVersion`，使陈旧行被丢弃，而不是被正向 apply 成垃圾。 +- **状态是纯 JSON，`stateVersion` 是其失效锚点。** 持久投影缓存（persisted projection cache）存储 `(sessionId, key, ver, seq, val)` 行；状态形状或折叠语义一旦变化就递增 `stateVersion`，使陈旧行被丢弃，而不是被正向 apply 成垃圾。 - **本层没有协议词汇。** 注册表只暴露变更流与快照读取面；载体（api-proxy）据此自铸各自的帧（`session/projection`）与块。 - **可选 seam。** 领域插件在 `ctx.inject(['sessionProjections'], …)` 下注册，因此不带注册表的 headless 组装完全不受影响；载体使用 `ctx.get('sessionProjections')`，注册表缺席时完全省略自己的块与帧。 @@ -43,5 +43,5 @@ - **每个尾页携带每个已注册的 key**——尚无逐 key 的 opt-out 或惰性 key 请求形状；在值都是 UI 量级的全量状态（一张 todo 清单、一份 goal 快照）时可以接受，若某领域的值变大再重议。 - **正向驱动（eager drive）逐事件触达每个单元**——按构造开销很低（全量值规则、同引用闸门），但若出现热点路径，可加按单元的事件类型预过滤，契约不变。 -- **持久投影缓存属于后续阶段**——cell 目前只活在内存里；重启后首次触达时靠折叠内存日志重建。`stateVersion` 字段是为该阶段预先声明的失效锚点。 +- **注册表 cell 只活在内存里**——重启后首次触达时靠折叠日志重建；挂载了 `dsh-session-projection-cache` 的组合改由持久行播种该折叠。 - **单元同步纪律只有部分可机械把关**——边界 `schema.parse` 能拒绝返回 Promise 的 `view`，但阻塞的 `apply`、或读取撕裂的非会话状态的 `apply`，只能靠评审把关；invariant 配套记载了为何不存在运行时检查。 diff --git a/packages/session-projection/session-projection/src/index.ts b/packages/session-projection/session-projection/src/index.ts index 4e5274b463..dbf0c5ee6e 100644 --- a/packages/session-projection/session-projection/src/index.ts +++ b/packages/session-projection/session-projection/src/index.ts @@ -66,9 +66,9 @@ export interface ProjectionDefinition { view(state: S): SessionProjectionMap[K] /** * Persisted-cache invalidation anchor: bump whenever the state shape or the - * fold semantics change, so persisted `(sessionId, key, stateVersion, - * observedSeq, state)` rows from an older unit are discarded instead of - * being forward-applied into garbage. Non-negative integer. + * fold semantics change, so persisted `(sessionId, key, ver, seq, val)` + * rows from an older unit are discarded instead of being forward-applied + * into garbage. Non-negative integer. */ stateVersion: number } @@ -99,20 +99,19 @@ export interface ProjectionSnapshot { /** * One unit's checkpoint: its internal state (plain JSON by the unit - * contract), the seq of the last event folded into it, and the + * contract), the seq of the last event folded into it, and the unit * `stateVersion` that produced it — the persisted projection-cache row - * `(sessionId, key, stateVersion, observedSeq, state)` minus the two outer - * keys. A row is never authoritative, only a fold shortcut: `restore` - * discards it on a `stateVersion` mismatch or when it claims events past the - * stored log end. + * `(sessionId, key, ver, seq, val)` minus the two outer keys. A row is + * never authoritative, only a fold shortcut: `restore` discards it on a + * version mismatch or when it claims events past the stored log end. */ export interface ProjectionCheckpointRow { /** The registering unit's `stateVersion` at fold time. */ - stateVersion: number - /** Seq of the last event folded into `state`; -1 for the empty log. */ - observedSeq: number + ver: number + /** Seq of the last event folded into `val`; -1 for the empty log. */ + seq: number /** The unit's internal state — plain JSON per the unit contract. */ - state: unknown + val: unknown } /** Checkpoint rows keyed by projection key (one session's persisted cache value). */ @@ -231,9 +230,9 @@ export class SessionProjectionRegistry extends Service { * State-level checkpoint of every registered unit for one session, read * from the watermark cache (missing cells fold lazily over the in-memory * log). This is the write side of the persisted projection cache: the - * returned rows are the `(key → {stateVersion, observedSeq, state})` part - * of the durable `(sessionId, key, stateVersion, observedSeq, state)` - * rows. Every `state` is a DETACHED structured clone — never the live + * returned rows are the `(key → {ver, seq, val})` part of the durable + * `(sessionId, key, ver, seq, val)` + * rows. Every `val` is a DETACHED structured clone — never the live * cell reference: the watermark cache is this registry's authoritative * mutable state, and a caller reaching the live reference could corrupt * every subsequent snapshot and frame through it (plain JSON by the unit @@ -246,9 +245,9 @@ export class SessionProjectionRegistry extends Service { for (const registration of this.registrations.values()) { const cell = this.cellFor(registration, session) rows[registration.def.key] = { - stateVersion: registration.def.stateVersion, - observedSeq: cell.observedSeq, - state: structuredClone(cell.state), + ver: registration.def.stateVersion, + seq: cell.observedSeq, + val: structuredClone(cell.state), } } return rows @@ -257,7 +256,7 @@ export class SessionProjectionRegistry extends Service { /** * The stored seq a {@link restore} tail read over `checkpoint` must start * at: one event BELOW the lowest usable watermark (a row is usable when - * its `stateVersion` matches the live unit; an absent or mismatched row + * its `ver` matches the live unit's `stateVersion`; an absent or mismatched row * pulls the floor to `0` — that key must refold the full log). The * one-below anchor is load-bearing: the tail then proves how far the * stored log still extends, so {@link restore} can detect a log that @@ -274,8 +273,8 @@ export class SessionProjectionRegistry extends Service { let floor: number | undefined for (const registration of this.registrations.values()) { const row = checkpoint[registration.def.key] - const need = row !== undefined && row.stateVersion === registration.def.stateVersion - ? Math.max(row.observedSeq + 1, 0) + const need = row !== undefined && row.ver === registration.def.stateVersion + ? Math.max(row.seq + 1, 0) : 0 floor = floor === undefined ? need : Math.min(floor, need) } @@ -284,7 +283,7 @@ export class SessionProjectionRegistry extends Service { /** * View a checkpoint's rows without any log read: for every registered - * unit whose row's `stateVersion` matches, serve the schema-validated + * unit whose row's `ver` matches, serve the schema-validated * `view` of the stored state; mismatched or absent rows leave their key * absent (a cold or listing consumer treats it as not-yet-available and a * fuller read path refolds it). The zero-I/O rung of the read ladder — @@ -297,8 +296,8 @@ export class SessionProjectionRegistry extends Service { for (const registration of this.registrations.values()) { const def = registration.def const row = checkpoint[def.key] - if (row === undefined || row.stateVersion !== def.stateVersion) continue - values[def.key] = def.schema.parse(def.view(row.state)) + if (row === undefined || row.ver !== def.stateVersion) continue + values[def.key] = def.schema.parse(def.view(row.val)) } return values } @@ -311,9 +310,9 @@ export class SessionProjectionRegistry extends Service { * `readFrom(id, restoreFloor(checkpoint))` and that same floor as * `baseSeq`; the floor's one-below anchor makes the supplied end honest, * so a shrunk log is detected here. A row is usable iff its - * `stateVersion` matches the live unit, it does not predate `baseSeq` - * (`observedSeq >= baseSeq - 1`), and it does not claim events past the - * supplied end (`observedSeq <= endSeq`); an unusable row is discarded + * `ver` matches the live unit's `stateVersion`, it does not predate `baseSeq` + * (`seq >= baseSeq - 1`), and it does not claim events past the + * supplied end (`seq <= endSeq`); an unusable row is discarded * and its key refolds from `init` — which is only sound over the full * log, so a discarded row with `baseSeq > 0` throws (the caller re-reads * from seq 0, e.g. after a crash-repair truncation shrank the log below @@ -334,22 +333,22 @@ export class SessionProjectionRegistry extends Service { const def = registration.def const row = checkpoint[def.key] const usable = row !== undefined - && row.stateVersion === def.stateVersion - && row.observedSeq >= baseSeq - 1 - && row.observedSeq <= endSeq + && row.ver === def.stateVersion + && row.seq >= baseSeq - 1 + && row.seq <= endSeq if (!usable && baseSeq > 0) { throw new Error( `session projection ${JSON.stringify(def.key)} cannot restore from seq ${baseSeq}: ` + 'its checkpoint row is missing, version-mismatched, or beyond the supplied log end; re-read from seq 0', ) } - let state = usable ? row.state : def.init() - const from = usable ? row.observedSeq : baseSeq - 1 + let state = usable ? row.val : def.init() + const from = usable ? row.seq : baseSeq - 1 for (const event of events) { if (event.seq > from) state = def.apply(state, event) } values[def.key] = def.schema.parse(def.view(state)) - refreshed[def.key] = { stateVersion: def.stateVersion, observedSeq: endSeq, state } + refreshed[def.key] = { ver: def.stateVersion, seq: endSeq, val: state } } return { snapshot: { asOfSeq: endSeq, values: values }, diff --git a/packages/session-projection/session-projection/tests/registry.spec.ts b/packages/session-projection/session-projection/tests/registry.spec.ts index 2f2ab15dde..90b8de4a40 100644 --- a/packages/session-projection/session-projection/tests/registry.spec.ts +++ b/packages/session-projection/session-projection/tests/registry.spec.ts @@ -175,11 +175,11 @@ describe('SessionProjectionRegistry drive', () => { ctx.sessionProjections.register({ ...countUnit(), stateVersion: 7 }) const markEvent = mark(session, ['a']) const rows = ctx.sessionProjections.checkpoint(session) - expect(rows['test/marks']).toEqual({ stateVersion: 1, observedSeq: markEvent.seq, state: { marks: ['a'] } }) - expect(rows['test/count']).toEqual({ stateVersion: 7, observedSeq: markEvent.seq, state: 1 }) + expect(rows['test/marks']).toEqual({ ver: 1, seq: markEvent.seq, val: { marks: ['a'] } }) + expect(rows['test/count']).toEqual({ ver: 7, seq: markEvent.seq, val: 1 }) // Empty log: init-derived state at watermark -1. const fresh = ctx.sessions.create() - expect(ctx.sessionProjections.checkpoint(fresh)['test/marks']).toEqual({ stateVersion: 1, observedSeq: -1, state: null }) + expect(ctx.sessionProjections.checkpoint(fresh)['test/marks']).toEqual({ ver: 1, seq: -1, val: null }) }) it('checkpoint states are detached clones — mutating them cannot corrupt the watermark cache', async () => { @@ -188,11 +188,11 @@ describe('SessionProjectionRegistry drive', () => { mark(session, ['a']) const rows = ctx.sessionProjections.checkpoint(session) // Hostile (or merely careless) consumer mutates the handed-out state. - ;(rows['test/marks']?.state as { marks: string[] }).marks.push('INJECTED') + ;(rows['test/marks']?.val as { marks: string[] }).marks.push('INJECTED') // The registry's authoritative cell is untouched: snapshot and a fresh // checkpoint both still serve the committed value. expect(ctx.sessionProjections.snapshot(session).values['test/marks']).toEqual({ marks: ['a'] }) - expect(ctx.sessionProjections.checkpoint(session)['test/marks']?.state).toEqual({ marks: ['a'] }) + expect(ctx.sessionProjections.checkpoint(session)['test/marks']?.val).toEqual({ marks: ['a'] }) }) it('restoreFloor anchors one below the lowest usable watermark and at 0 for missing or mismatched rows', async () => { @@ -204,18 +204,18 @@ describe('SessionProjectionRegistry drive', () => { // Lowest usable watermark is count's 5 → the anchored tail starts AT 5 // (one below the first needed seq 6), so the read proves seq 5 still exists. expect(ctx.sessionProjections.restoreFloor({ - 'test/marks': { stateVersion: 1, observedSeq: 10, state: { marks: [] } }, - 'test/count': { stateVersion: 1, observedSeq: 5, state: 6 }, + 'test/marks': { ver: 1, seq: 10, val: { marks: [] } }, + 'test/count': { ver: 1, seq: 5, val: 6 }, })).toBe(5) // A version-mismatched row forces that key back to a full refold. expect(ctx.sessionProjections.restoreFloor({ - 'test/marks': { stateVersion: 2, observedSeq: 10, state: { marks: [] } }, - 'test/count': { stateVersion: 1, observedSeq: 5, state: 6 }, + 'test/marks': { ver: 2, seq: 10, val: { marks: [] } }, + 'test/count': { ver: 1, seq: 5, val: 6 }, })).toBe(0) // A fresh (-1) row still needs the whole tail from 0. expect(ctx.sessionProjections.restoreFloor({ - 'test/marks': { stateVersion: 1, observedSeq: -1, state: null }, - 'test/count': { stateVersion: 1, observedSeq: -1, state: 0 }, + 'test/marks': { ver: 1, seq: -1, val: null }, + 'test/count': { ver: 1, seq: -1, val: 0 }, })).toBe(0) }) @@ -230,8 +230,8 @@ describe('SessionProjectionRegistry drive', () => { // marks row usable (watermark 2, tail starts at 3); count row mismatched — but // a mismatch with baseSeq > 0 cannot silently refold: it throws for a re-read. expect(() => ctx.sessionProjections.restore({ - 'test/marks': { stateVersion: 1, observedSeq: 2, state: { marks: ['old'] } }, - 'test/count': { stateVersion: 99, observedSeq: 2, state: 3 }, + 'test/marks': { ver: 1, seq: 2, val: { marks: ['old'] } }, + 'test/count': { ver: 99, seq: 2, val: 3 }, }, tail, 3)).toThrow(/re-read from seq 0/) // The full-log re-read (baseSeq 0) refolds the mismatched key from init. const full: SessionEvent[] = [ @@ -241,15 +241,15 @@ describe('SessionProjectionRegistry drive', () => { ...tail, ] const { snapshot, checkpoint } = ctx.sessionProjections.restore({ - 'test/marks': { stateVersion: 1, observedSeq: 2, state: { marks: ['old', '2'] } }, - 'test/count': { stateVersion: 99, observedSeq: 2, state: 3 }, + 'test/marks': { ver: 1, seq: 2, val: { marks: ['old', '2'] } }, + 'test/count': { ver: 99, seq: 2, val: 3 }, }, full, 0) expect(snapshot.asOfSeq).toBe(4) expect(snapshot.values['test/marks']).toEqual({ marks: ['new'] }) expect(snapshot.values['test/count']).toBe(5) // refolded from init over all 5 events // The refreshed rows sit at the served cut, ready for a durable write-back. - expect(checkpoint['test/marks']).toEqual({ stateVersion: 1, observedSeq: 4, state: { marks: ['new'] } }) - expect(checkpoint['test/count']).toEqual({ stateVersion: 1, observedSeq: 4, state: 5 }) + expect(checkpoint['test/marks']).toEqual({ ver: 1, seq: 4, val: { marks: ['new'] } }) + expect(checkpoint['test/count']).toEqual({ ver: 1, seq: 4, val: 5 }) }) it('restore over a suffix folds only past each row watermark and serves an exact empty-tail cut', async () => { @@ -257,8 +257,8 @@ describe('SessionProjectionRegistry drive', () => { ctx.sessionProjections.register(marksUnit()) ctx.sessionProjections.register(countUnit()) const rows = { - 'test/marks': { stateVersion: 1, observedSeq: 4, state: { marks: ['done'] } }, - 'test/count': { stateVersion: 1, observedSeq: 2, state: 3 }, + 'test/marks': { ver: 1, seq: 4, val: { marks: ['done'] } }, + 'test/count': { ver: 1, seq: 2, val: 3 }, } const tail: SessionEvent[] = [ { type: 'turn/start', seq: 3, time: 3, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } }, @@ -273,8 +273,8 @@ describe('SessionProjectionRegistry drive', () => { // Empty tail (checkpoint is current): the cut sits at baseSeq - 1. const { snapshot: current } = ctx.sessionProjections.restore({ - 'test/marks': { stateVersion: 1, observedSeq: 4, state: { marks: ['done'] } }, - 'test/count': { stateVersion: 1, observedSeq: 4, state: 5 }, + 'test/marks': { ver: 1, seq: 4, val: { marks: ['done'] } }, + 'test/count': { ver: 1, seq: 4, val: 5 }, }, [], 5) expect(current.asOfSeq).toBe(4) expect(current.values['test/count']).toBe(5) @@ -285,8 +285,8 @@ describe('SessionProjectionRegistry drive', () => { ctx.sessionProjections.register(marksUnit()) ctx.sessionProjections.register(countUnit()) const values = ctx.sessionProjections.viewCheckpoint({ - 'test/marks': { stateVersion: 1, observedSeq: 4, state: { marks: ['stored'] } }, - 'test/count': { stateVersion: 99, observedSeq: 4, state: 5 }, // mismatched: absent + 'test/marks': { ver: 1, seq: 4, val: { marks: ['stored'] } }, + 'test/count': { ver: 99, seq: 4, val: 5 }, // mismatched: absent }) expect(values['test/marks']).toEqual({ marks: ['stored'] }) expect('test/count' in values).toBe(false) @@ -296,7 +296,7 @@ describe('SessionProjectionRegistry drive', () => { it('restore rejects a row claiming events past the supplied log end (shrunk log ⇒ re-read)', async () => { const { ctx } = await harness() ctx.sessionProjections.register(countUnit()) - const rows = { 'test/count': { stateVersion: 1, observedSeq: 9, state: 10 } } + const rows = { 'test/count': { ver: 1, seq: 9, val: 10 } } // The anchored floor sits ON the watermark, so the tail read must return // at least seq 9 from an intact log… const floor = ctx.sessionProjections.restoreFloor(rows)