fix(session): centralize projection baseline precedence

.agents/notes/implemented/architecture/2026-08-25-session-observations-and-projection-owned-client-state.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write .agents/notes/implemented/architecture/2026-08-25-session-observations-and-projection-owned-client-state.md
-2026-08-25-session-observations-and-projection-owned-client-state.md: 70919f6543cf0751c9afa64df187dfc82c27dcd5
-2026-08-25-session-observations-and-projection-owned-client-state.zh.md: cba093287c2c91644b82af5b63f6d75e148a6a27
+2026-08-25-session-observations-and-projection-owned-client-state.md: 9bf4bb7c720c4fccbfc162ca5a3250c9a6fdb178
+2026-08-25-session-observations-and-projection-owned-client-state.zh.md: 0043ab237096f1533de383eb7d63bde02fe6cf3b

.agents/notes/implemented/architecture/2026-08-25-session-observations-and-projection-owned-client-state.md

@@ -108,11 +108,11 @@ These distinctions prevent one overloaded `undefined` from representing cache mi
 | Follow opening baseline | Complete for the Host composition | Exact opening cursor | Capability absent |
 | Projection frame | One whole key | Event sequence carried by the frame | Not applicable |
 
-The Client stores one `{ value, seq }` row per key. Partial list hints and ordinary projection frames use higher-sequence-wins, while a successful follow opening baseline is an exact replacement even when a tentative cache row claims a higher cut. During initial open, explicit resync, or carrier reconnection, the Session normalizes captured control input to the latest replacement baseline and only the frames that follow it. After installing the exact opening value, it applies that control baseline only when its durable cut is at least the opening cut, then applies subsequent frames only when they advance the selected complete cut. A replacement control baseline that remains authoritative first discards rows beyond its durable cut before seeding its complete values.
+The Client projection store records source provenance and arrival revision beside each `{ value, seq }` row. A list hint updates only tentative rows before the store has accepted a complete authoritative cut. The first authoritative frame replaces a tentative row regardless of sequence, while later frames require a strictly higher sequence. Follow and control baselines are complete replacements, so they replace equal-sequence values and remove omitted keys.
 
-The list view reads the same per-Session store as the opened Session. Hints can populate title, preset, and other list presentation before the first exact opening; after that baseline is installed, late list hints for the resident Session are ignored so tentative cache data cannot re-enter the opened value.
+Each Session retains one opaque token for initial open, explicit resync, or carrier reconnection and cancels it if that opening fails or is superseded. When a follow baseline completes the token, the store discards pre-token state and retains only authoritative frames that arrived after the token and are newer than the opening cut. A control baseline received after the token at an equal or newer cut remains authoritative. Session does not capture, buffer, or replay projection operations.
 
-The per-Session Client projection store never folds Session events. It keeps higher-sequence ordering for hints and whole-value frames, supports exact replacement for an authoritative follow baseline, and applies replacement-control truncation under the Session-owned reconnect replay boundary.
+The list view and opened Session read the same per-Session store. Hints can populate title, preset, and other list presentation before the first complete authoritative cut; afterward, late hints are ignored so tentative cache data cannot re-enter the opened value. The store never folds Session events: it owns hint/frame provenance, frame ordering, exact baseline replacement, and opening reconciliation; `Session` owns only token lifetime, and `SessionManager` routes list hints, control frames, and control baselines to that resident store.
 
 Data that is not derived from one Session remains outside projections. `llm.models` owns the Host-generation model catalog, and `agentPreset.list` owns the configurable preset roster. A selector combines the relevant catalog with the Session's `modelSelection` or `agentPreset` projection only when both inputs are ready. During refresh it may retain the last complete catalog; before the first complete pair it reports loading instead of rendering a guessed name or availability verdict.
 
@@ -146,8 +146,8 @@ Cancellation stops queued or in-flight cold resolution at documented checkpoints
 | Exact live-preferred read cut | SessionQuery observation | Individual endpoint helpers |
 | Fold state and client-value computation | Projection registry and domain unit | SessionQuery and Client |
 | Partial list acceleration | Projection cache and list policy | Follow protocol |
-| Opening and reconnect replacement | Session follow and journal stream | Session page |
-| Per-key value ordering | Client projection store | Domain UI components |
+| Opening and reconnect token lifetime | Client Session | Session page and domain UI components |
+| Hint, frame, and complete-baseline precedence | Client projection store | Client Session and domain UI components |
 | Provider or preset catalog lifecycle | Its catalog directory | Session projection |
 | Rendering and transient interaction state | Domain UI package | Host projection units |
 
@@ -174,7 +174,7 @@ These rules apply to new Session-derived Client state even when a direct event s
 
 Persistence and SessionQuery tests pin shared cold loading, cancellation, live-source races, retained observations, disposal, and all-or-none projection calculation. Session Controller and Gateway tests pin snapshot-first opening, replacement reconnect, older-page reads, gap repair, list-cache hints, bounded small-log fallback, and promotion after snapshot delivery.
 
-Client tests pin higher-sequence-wins projection storage, title updates, model catalog and selection readiness, preset roster refresh and Session-specific selection, and subagent loading without transient offline presentation. Subagent tests pin corpus enumeration, cache and observation fallback, lifecycle witnesses, bounded cold reads, and no Agent activation during listing.
+Client tests pin tentative-hint ordering, first-authoritative-frame takeover, exact opening and equal-cut control replacement, post-token frame retention, stale or canceled opening tokens, manager/Session shared-store routing, title updates, model catalog and selection readiness, preset roster refresh and Session-specific selection, and subagent loading without transient offline presentation. Subagent tests pin corpus enumeration, cache and observation fallback, lifecycle witnesses, bounded cold reads, and no Agent activation during listing.
 
 ## Alternatives considered
 

.agents/notes/implemented/architecture/2026-08-25-session-observations-and-projection-owned-client-state.zh.md

@@ -108,11 +108,11 @@ Projection 的三种交付状态含义不同：
 | Follow opening baseline | 对当前 Host composition 完整 | 精确 opening cursor | Capability 不存在 |
 | Projection frame | 单个完整 key | Frame 携带的 event sequence | 不适用 |
 
-Client 为每个 key 保存一条 `{ value, seq }` row。部分 list hint 与普通 projection frame 遵循 seq 高者胜，而成功的 follow opening baseline 是精确替换，即使暂存 cache row 声称更高 cut 也一样。初次打开、显式 resync 或 carrier 重连期间，Session 会把捕获的 control 输入归一为最后一份 replacement baseline 及其后到达的 frame。安装精确 opening 值后，只有该 control baseline 的 durable cut 不早于 opening cut 时才应用它，随后也只应用能够推进所选完整 cut 的 frame。仍具权威性的 replacement control baseline 会先丢弃超出自身 durable cut 的 row，再播种完整值。
+Client projection store 在每条 `{ value, seq }` row 旁记录来源类别与到达 revision。list hint 只会在 store 尚未接收完整权威 cut 时更新暂定 row。首个权威 frame 无论 sequence 如何都会替换暂定 row，后续 frame 则必须具有严格更高的 sequence。follow 与 control baseline 都是完整替换，因此会覆盖等 sequence 值并移除缺失 key。
 
-List view 与已打开 Session 读取同一个 per-Session store。首次精确 opening 前，hint 可以填充 title、preset 和其他 list presentation；该 baseline 安装后，resident Session 会忽略迟到的 list hint，避免暂存 cache 数据重新进入已打开值。
+每个 Session 为初次打开、显式 resync 或 carrier 重连保留一个不透明 token，并在该 opening 失败或被取代时取消它。当 follow baseline 完成该 token 时，store 丢弃 token 之前的状态，只保留 token 之后到达且 seq 新于 opening cut 的权威 frame。若 control baseline 在 token 之后到达且 cut 等于或新于 opening cut，它保持权威。Session 不捕获、缓冲或重放 projection operation。
 
-每个 Session 的 Client projection store 从不折叠 Session event。它对 hint 与 whole-value frame 保持 seq 排序，为权威 follow baseline 提供精确替换，并在 Session 拥有的重连重放边界内应用 replacement-control 截断。
+List view 与已打开 Session 读取同一个 per-Session store。首次完整权威 cut 之前，hint 可以填充 title、preset 和其他 list presentation；之后迟到的 hint 会被忽略，避免暂定 cache 数据重新进入已打开值。store 从不折叠 Session event：它拥有 hint/frame 来源、frame 排序、完整 baseline 的精确替换和 opening reconciliation；`Session` 只拥有 token 生命周期，`SessionManager` 则把 list hint、control frame 与 control baseline 路由到这份 resident store。
 
 不由单个 Session 派生的数据不进入 projection。`llm.models` 拥有当前 Host generation 的 model catalog，`agentPreset.list` 拥有可配置 preset roster。Selector 只在相应 catalog 与 Session 的 `modelSelection` 或 `agentPreset` projection 均就绪后组合两者。刷新时可以保留上一份完整 catalog；第一次获得完整输入前显示 loading，而不是展示猜测的名称或可用性结论。
 
@@ -146,8 +146,8 @@ Client 本地交互状态也继续留在本地：loading 和 error 状态、打
 | 精确 live-preferred read cut | SessionQuery observation | 各 endpoint helper |
 | Fold state 与 Client-value 计算 | Projection registry 与 domain unit | SessionQuery 与 Client |
 | 部分 list acceleration | Projection cache 与 list policy | Follow protocol |
-| Opening 与 reconnect replacement | Session follow 与 journal stream | Session page |
-| Per-key value ordering | Client projection store | Domain UI component |
+| Opening 与 reconnect token 生命周期 | Client Session | Session page 与 domain UI component |
+| Hint、frame 与完整 baseline 的 precedence | Client projection store | Client Session 与 domain UI component |
 | Provider 或 preset catalog lifecycle | 对应 catalog directory | Session projection |
 | Rendering 与瞬时 interaction state | Domain UI package | Host projection unit |
 
@@ -174,7 +174,7 @@ Client 本地交互状态也继续留在本地：loading 和 error 状态、打
 
 Persistence 与 SessionQuery 测试固定共享冷加载、取消、live-source race、retained observation、dispose 和 all-or-none projection 计算。Session Controller 与 Gateway 测试固定 snapshot-first opening、replacement reconnect、旧分页读取、gap repair、list-cache hints、小日志有界 fallback，以及 snapshot 交付后的 promotion。
 
-Client 测试固定 higher-sequence-wins projection store、title 更新、model catalog 与 selection readiness、preset roster refresh 与 Session 专属选择，以及不会短暂展示离线状态的 subagent loading。Subagent 测试固定 corpus 枚举、cache 与 observation fallback、lifecycle witness、有界冷读，以及 listing 期间不激活 Agent。
+Client 测试固定暂定 hint 排序、首个权威 frame 接管、opening 与等 cut control 精确替换、token 后 frame 保留、陈旧或已取消的 opening token、manager/Session 共用 store、title 更新、model catalog 与 selection readiness、preset roster refresh 与 Session 专属选择，以及不会短暂展示离线状态的 subagent loading。Subagent 测试固定 corpus 枚举、cache 与 observation fallback、lifecycle witness、有界冷读，以及 listing 期间不激活 Agent。
 
 ## 考虑过的替代方案
 

packages/api/session-controller/src/client/sessions/manager.ts

@@ -488,21 +488,13 @@ export class SessionManager {
             session.handleBlank(s.blank)
             session.handleRunning(s.running)
           }
-          // Apply each row's projection values (cold values surface without
-          // opening the session). The list block is partial, so an absent key
-          // must not clear. Once a resident Session has installed its exact
-          // opening baseline, it ignores later tentative list hints.
+          // Apply each row's tentative projection values (cold values surface
+          // without opening the session). The store owns hint precedence and
+          // ignores them after a complete authoritative baseline.
           for (const s of result.value.items) {
             const block = s.projections
             if (block === undefined) continue
-            const session = this.sessions.get(s.sessionId)
-            if (session !== undefined) {
-              session.handleProjectionHint(block)
-              continue
-            }
-            const store = this.projectionStore(s.sessionId)
-            const values = block.values as Record<string, unknown>
-            for (const key of Object.keys(values)) store.apply(key, values[key], block.asOfSeq)
+            this.projectionStore(s.sessionId).prewarm(block)
           }
         } else {
           this.listState = 'error'
@@ -678,9 +670,7 @@ export class SessionManager {
       return
     }
     if (frame.type === 'projection') {
-      const session = this.sessions.get(frame.sessionId)
-      if (session === undefined) this.projectionStore(frame.sessionId).apply(frame.key, frame.value, frame.seq)
-      else session.handleProjectionFrame(frame)
+      this.projectionStore(frame.sessionId).apply(frame.key, frame.value, frame.seq)
       this.notifier.markDirty()
       return
     }
@@ -706,15 +696,7 @@ export class SessionManager {
     }
 
     for (const [sessionId, block] of Object.entries(baseline.projections)) {
-      const id = sessionId as SessionId
-      const session = this.sessions.get(id)
-      if (session !== undefined) {
-        session.replaceProjectionBaseline(block)
-        continue
-      }
-      const store = this.projectionStore(id)
-      store.truncate(block.asOfSeq)
-      store.seed(block)
+      this.projectionStore(sessionId as SessionId).replaceControlBaseline(block)
     }
     for (const [sessionId, session] of this.sessions) {
       session.replaceControl(this.queues.get(sessionId) ?? [])
@@ -730,17 +712,7 @@ export class SessionManager {
     this.mergeSummary(summary)
     this.sessions.get(summary.sessionId)?.handleBlank(summary.blank)
     const projections = summary.projections
-    if (projections !== undefined) {
-      const session = this.sessions.get(summary.sessionId)
-      if (session !== undefined) {
-        session.handleProjectionHint(projections)
-      } else {
-        const store = this.projectionStore(summary.sessionId)
-        for (const [key, value] of Object.entries(projections.values)) {
-          store.apply(key, value, projections.asOfSeq)
-        }
-      }
-    }
+    if (projections !== undefined) this.projectionStore(summary.sessionId).prewarm(projections)
     if (summary.origin === 'subagent' && summary.parentSessionId !== undefined) {
       this.markCatalogParentExpandable(summary.parentSessionId)
     }

packages/api/session-controller/src/client/sessions/projection-store.ts

@@ -2,12 +2,11 @@
  * Generic per-session projection value store (push model; see the
  * session-projection subsystem page, docs/subsystems/session-projection.md):
  * the host is the only computation site; the client holds finished
- * whole values per key — `key → { value, seq }` — seeded by Session-list and
- * session-added hints, exactly replaced by a successful follow opening, and
- * updated by control baselines and Session Controller `projection` frames.
- * Ordinary updates use **higher seq wins**. No client-side domain folding exists: a domain ships projection
- * support with zero client code. Per-key bare observable faces feed
- * `useProjection` (ui-renderer binds them).
+ * whole values per key. The store distinguishes tentative list hints from
+ * authoritative frames and complete baselines, and owns their precedence
+ * across opening and reconnect lifecycles. No client-side domain folding
+ * exists: a domain ships projection support with zero client code. Per-key
+ * bare observable faces feed `useProjection` (ui-renderer binds them).
  */
 import type { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection/types'
 import type { ObservableSnapshot } from '@deepseek-ai/dsh-client-store'
@@ -52,10 +51,18 @@ export interface ProjectionsBaseline {
   values: Readonly<Record<string, unknown>>
 }
 
-/** One key's row: the latest finished value and the seq it is consistent with. */
+/** Opaque marker for one exact Session opening lifecycle. */
+export interface ProjectionOpeningToken {
+  /** Store revision when the opening began. */
+  readonly revision: number
+}
+
+/** One key's row with its authority and arrival revision. */
 interface Row {
   value: unknown
   seq: number
+  provenance: 'tentative' | 'authoritative'
+  revision: number
 }
 
 /** Per-key notification channel: the bare face plus its batching notifier. */
@@ -66,18 +73,21 @@ interface Channel {
 
 /**
  * One session's projection values. Framework semantics are uniform across
- * ordinary source: a partial list block applies its carried keys, a control
- * baseline clears omitted keys at its cut, and a push frame updates one row.
- * The owning Session separately uses {@link replace} for an authoritative
- * follow opening. A key the store has never seen
- * reads `undefined` (capability absent). Faces are identity-stable per key
- * (create-on-demand, cached) so the React side binds each exactly once; the
- * store-level channel (`subscribeAny`) serves coarse consumers.
+ * source: list hints are tentative, frames are authoritative per-key updates,
+ * and opening/control baselines are complete authoritative cuts. A key the
+ * store has never seen reads `undefined` (capability absent). Faces are
+ * identity-stable per key (create-on-demand, cached) so the React side binds
+ * each exactly once; the store-level channel (`subscribeAny`) serves coarse
+ * consumers.
  */
 export class ProjectionValueStore {
   private readonly rows = new Map<string, Row>()
   private readonly channels = new Map<string, Channel>()
   private valuesCache: Readonly<Partial<SessionProjectionMap>> | undefined
+  private revision = 0
+  private activeOpening: ProjectionOpeningToken | undefined
+  private latestControlBaseline: { readonly revision: number; readonly asOfSeq: number } | undefined
+  private completeBaselineInstalled = false
   /** Coarse any-key channel (no snapshot cache to rebuild: reads hit rows directly). */
   private readonly anyNotifier = new Notifier(() => {})
 
@@ -126,72 +136,102 @@ export class ProjectionValueStore {
   }
 
   /**
-   * Apply one finished value from the Session control stream.
+   * Apply a partial cache-backed hint while no complete authoritative cut exists.
+   * @param hint - partial projection values from the Session list cache.
+   */
+  prewarm(hint: ProjectionsBaseline): void {
+    if (this.completeBaselineInstalled) return
+    const values = hint.values as Record<string, unknown>
+    for (const key of Object.keys(values)) {
+      const previous = this.rows.get(key)
+      if (previous?.provenance === 'authoritative') continue
+      if (previous !== undefined && hint.asOfSeq <= previous.seq) continue
+      this.rows.set(key, {
+        value: values[key],
+        seq: hint.asOfSeq,
+        provenance: 'tentative',
+        revision: ++this.revision,
+      })
+      this.changed(key)
+    }
+  }
+
+  /**
+   * Apply one authoritative finished value from the Session control stream.
+   * The first frame replaces a tentative row regardless of sequence; later
+   * authoritative frames use strict higher-sequence ordering.
    * @param key - projection key.
    * @param value - whole value computed by the host unit.
    * @param seq - the unit's watermark at emission.
    */
   apply(key: string, value: unknown, seq: number): void {
     const row = this.rows.get(key)
-    if (row !== undefined && seq <= row.seq) return
-    this.rows.set(key, { value, seq })
+    if (row?.provenance === 'authoritative' && seq <= row.seq) return
+    this.rows.set(key, {
+      value,
+      seq,
+      provenance: 'authoritative',
+      revision: ++this.revision,
+    })
     this.changed(key)
   }
 
   /**
-   * Seed from a complete history or control projections block. Every carried
-   * key lands under the same higher-seq-wins rule as frames. A key the block
-   * omits is capability-absent as of the cut, so its row clears unless a newer
-   * value already superseded the baseline.
-   * @param baseline - the response's projections block.
+   * Start one exact opening. The token lets completion distinguish state that
+   * existed before the request from authoritative frames arriving afterward.
+   * @returns an opaque token owned by the caller's opening lifecycle.
    */
-  seed(baseline: ProjectionsBaseline): void {
-    // Erased walk: the framework crosses the open key space; per-key typing
-    // is re-established at the consumer (useProjection's map lookup).
-    const values = baseline.values as Record<string, unknown>
-    for (const key of Object.keys(values)) this.apply(key, values[key], baseline.asOfSeq)
-    for (const [key, row] of this.rows) {
-      if (Object.hasOwn(values, key)) continue
-      if (row.seq > baseline.asOfSeq) continue
-      this.rows.delete(key)
-      this.changed(key)
-    }
+  beginOpening(): ProjectionOpeningToken {
+    const token = Object.freeze({ revision: this.revision })
+    this.activeOpening = token
+    return token
   }
 
   /**
-   * Install an authoritative complete baseline exactly, regardless of rows
-   * previously supplied by tentative cache hints or an earlier stream
-   * generation.
-   * @param baseline - the opening response's complete projections block.
+   * Complete an exact opening. A control baseline received after the token
+   * wins at an equal or newer cut. Otherwise the opening replaces all prior
+   * state and retains only authoritative post-token frames newer than its cut.
+   * @param token - token returned by {@link beginOpening}.
+   * @param baseline - complete projection values at the opening cut.
    */
-  replace(baseline: ProjectionsBaseline): void {
-    const values = baseline.values as Record<string, unknown>
-    const keys = new Set([...this.rows.keys(), ...Object.keys(values)])
-    for (const key of keys) {
-      if (!Object.hasOwn(values, key)) {
-        if (this.rows.delete(key)) this.changed(key)
-        continue
-      }
-      const value = values[key]
-      const previous = this.rows.get(key)
-      this.rows.set(key, { value, seq: baseline.asOfSeq })
-      if (previous === undefined || !Object.is(previous.value, value)) this.changed(key)
+  completeOpening(token: ProjectionOpeningToken, baseline: ProjectionsBaseline): void {
+    if (this.activeOpening !== token) return
+    this.activeOpening = undefined
+    const control = this.latestControlBaseline
+    if (control !== undefined
+      && control.revision > token.revision
+      && control.asOfSeq >= baseline.asOfSeq) {
+      return
     }
+
+    const retained = [...this.rows].filter(([, row]) =>
+      row.provenance === 'authoritative'
+      && row.revision > token.revision
+      && row.seq > baseline.asOfSeq)
+    this.completeBaselineInstalled = true
+    this.installCompleteBaseline(baseline, ++this.revision)
+    for (const [key, row] of retained) this.installRow(key, row)
   }
 
   /**
-   * Drop rows beyond a replacement control baseline. Such rows describe
-   * process state the Host lost before persisting it and would otherwise
-   * outrank recomputed lower-seq values forever. The caller seeds the new
-   * baseline immediately afterward.
-   * @param lastSeq - highest durable sequence reflected by the baseline.
+   * End one failed or superseded opening without changing the already visible
+   * values. Later openings receive a fresh revision boundary.
+   * @param token - token returned by {@link beginOpening}.
    */
-  truncate(lastSeq: number): void {
-    for (const [key, row] of this.rows) {
-      if (row.seq <= lastSeq) continue
-      this.rows.delete(key)
-      this.changed(key)
-    }
+  cancelOpening(token: ProjectionOpeningToken): void {
+    if (this.activeOpening === token) this.activeOpening = undefined
+  }
+
+  /**
+   * Replace the previous control-stream generation exactly, including rows at
+   * the same sequence. Frames arriving afterward again use higher-seq-wins.
+   * @param baseline - complete projections for the new control generation.
+   */
+  replaceControlBaseline(baseline: ProjectionsBaseline): void {
+    const revision = ++this.revision
+    this.completeBaselineInstalled = true
+    this.latestControlBaseline = { revision, asOfSeq: baseline.asOfSeq }
+    this.installCompleteBaseline(baseline, revision)
   }
 
   private changed(key: string): void {
@@ -200,6 +240,32 @@ export class ProjectionValueStore {
     this.anyNotifier.markDirty()
   }
 
+  /** Replace every row with one complete authoritative baseline. */
+  private installCompleteBaseline(baseline: ProjectionsBaseline, revision: number): void {
+    const values = baseline.values as Record<string, unknown>
+    const keys = new Set([...this.rows.keys(), ...Object.keys(values)])
+    for (const key of keys) {
+      if (!Object.hasOwn(values, key)) {
+        this.rows.delete(key)
+        this.changed(key)
+        continue
+      }
+      this.installRow(key, {
+        value: values[key],
+        seq: baseline.asOfSeq,
+        provenance: 'authoritative',
+        revision,
+      })
+    }
+  }
+
+  /** Install one row while notifying only when its observable value changes. */
+  private installRow(key: string, row: Row): void {
+    const previous = this.rows.get(key)
+    this.rows.set(key, row)
+    if (previous === undefined || !Object.is(previous.value, row.value)) this.changed(key)
+  }
+
   private channel(key: string): Channel {
     let channel = this.channels.get(key)
     if (channel === undefined) {

packages/api/session-controller/src/client/sessions/session.ts

@@ -34,7 +34,7 @@ import { Notifier } from './notifier.ts'
 import type { RemoteResult } from '@deepseek-ai/dsh-typert-protocol'
 import type { SessionRemotes } from './remotes.ts'
 import { ProjectionValueStore } from './projection-store.ts'
-import type { ProjectionsBaseline } from './projection-store.ts'
+import type { ProjectionOpeningToken, ProjectionsBaseline } from './projection-store.ts'
 import { resolvedClientTimeZone } from '../time-zone.ts'
 import { SessionQueueMirror } from './queue-mirror.ts'
 
@@ -64,14 +64,6 @@ export interface SessionOptions {
   projections?: ProjectionValueStore
 }
 
-type ProjectionFrame = Extract<SessionControlFrame, { type: 'projection' }>
-
-interface ProjectionCapture {
-  readonly generation: number
-  baseline?: ProjectionsBaseline
-  readonly frames: ProjectionFrame[]
-}
-
 /**
  * Owns a session's event window, lifecycle state, and observable
  * snapshot. React bindings remain outside this data layer. Features see only
@@ -88,10 +80,8 @@ export class Session implements SessionFace {
   /** Bumped by stream replacement to invalidate an in-flight doOpen. Stale
    *  passes drop all writes once the generation moves on. */
   private openGeneration = 0
-  /** Whether an authoritative event-stream projection baseline has replaced cache hints. */
-  private exactProjectionBaselineInstalled = false
-  /** Control operations that must be replayed after the current exact opening baseline. */
-  private projectionCapture: ProjectionCapture | undefined
+  /** Store-owned reconciliation token for the current exact opening. */
+  private projectionOpening: ProjectionOpeningToken | undefined
   private loadingOlder = false
   /** Authoritative stream-only inbox snapshot; pending work never hits history. */
   private readonly queueMirror = new SessionQueueMirror()
@@ -117,11 +107,9 @@ export class Session implements SessionFace {
   /**
    * Per-session projection value store (push model; see the session-projection
    * subsystem page, docs/subsystems/session-projection.md): finished whole
-   * values computed on the Host. Partial list hints and Session Controller
-   * frames use higher-seq-wins; a successful tail-page opening replaces those
-   * tentative rows exactly, then replays control operations received while it
-   * was in flight. Keys are
-   * read via `projections.faceOf(key)`
+   * values computed on the Host. The store owns precedence among tentative
+   * list hints, authoritative frames, complete control baselines, and exact
+   * opening baselines. Keys are read via `projections.faceOf(key)`
    * (the useProjection resolution face); the conversation snapshot never
    * carries projection values, and no client-side domain folding exists.
    * Manager-owned when constructed through SessionManager (frames route and
@@ -356,13 +344,7 @@ export class Session implements SessionFace {
     try {
       const result = toSessionResult(await this.remote.session.rename({ sessionId: this.sessionId, title }))
       if (result.ok) {
-        this.handleProjectionFrame({
-          type: 'projection',
-          sessionId: this.sessionId,
-          key: 'title',
-          value: result.value.title,
-          seq: result.value.seq,
-        })
+        this.projections.apply('title', result.value.title, result.value.seq)
       }
       return result
     } catch (error) {
@@ -387,7 +369,7 @@ export class Session implements SessionFace {
   open(): Promise<void> {
     if (this.openState === 'open') return Promise.resolve()
     if (this.openPromise !== null) return this.openPromise
-    this.beginProjectionCapture(this.openGeneration)
+    this.beginProjectionOpening()
     const promise = this.doOpen(this.openGeneration).finally(() => {
       // Identity-guarded: a superseded open must not null out the promise resync just started.
       if (this.openPromise === promise) this.openPromise = null
@@ -421,7 +403,8 @@ export class Session implements SessionFace {
   async resync(): Promise<void> {
     if (this.openState === 'cold') return // never opened: no window to rebuild (doOpen flips to 'loading' synchronously, so cold implies no in-flight open)
     this.openGeneration++
-    this.beginProjectionCapture(this.openGeneration)
+    this.cancelProjectionOpening()
+    this.beginProjectionOpening()
     const events = this.events
     this.events = undefined
     await events?.dispose()
@@ -473,35 +456,6 @@ export class Session implements SessionFace {
     this.notifier.markDirty()
   }
 
-  /**
-   * Apply and, while an exact opening replacement is pending, retain one live projection frame.
-   * @param frame - one live projection update for this Session.
-   */
-  handleProjectionFrame(frame: ProjectionFrame): void {
-    this.projections.apply(frame.key, frame.value, frame.seq)
-    this.captureProjectionFrame(frame)
-  }
-
-  /**
-   * Apply and retain one complete control-stream projection replacement.
-   * @param baseline - the complete projection baseline carried by the control stream.
-   */
-  replaceProjectionBaseline(baseline: ProjectionsBaseline): void {
-    this.applyProjectionBaseline(baseline)
-    this.captureProjectionBaseline(baseline)
-  }
-
-  /**
-   * Apply a tentative list/session-added cache hint until this Session has
-   * installed an exact event-stream baseline. Hints never join opening replay.
-   * @param baseline - a partial cache-backed projection hint.
-   */
-  handleProjectionHint(baseline: ProjectionsBaseline): void {
-    if (this.exactProjectionBaselineInstalled) return
-    const values = baseline.values as Record<string, unknown>
-    for (const key of Object.keys(values)) this.projections.apply(key, values[key], baseline.asOfSeq)
-  }
-
   /**
    * Running-bit relay from the host stream (list entry and snapshot stay consistent).
    * @param running - the new running state.
@@ -580,7 +534,7 @@ export class Session implements SessionFace {
    */
   async dispose(): Promise<void> {
     this.openGeneration++
-    this.projectionCapture = undefined
+    this.cancelProjectionOpening()
     const events = this.events
     this.events = undefined
     await events?.dispose()
@@ -596,11 +550,11 @@ export class Session implements SessionFace {
     const events = new SessionEventStream(this.remote, this.sessionAddress(), {
       publish: (change) => {
         if (generation !== this.openGeneration || this.events !== events) return
-        this.acceptEventChange(change, generation)
+        this.acceptEventChange(change)
       },
       carrierFailed: () => {
         if (generation !== this.openGeneration || this.events !== events) return
-        this.beginProjectionCapture(generation)
+        this.beginProjectionOpening()
       },
       failed: (error) => {
         this.failEventStream(events, generation, error)
@@ -614,7 +568,7 @@ export class Session implements SessionFace {
     } catch (error) {
       if (generation !== this.openGeneration || this.events !== events) return
       this.events = undefined
-      this.discardProjectionCapture(generation)
+      this.cancelProjectionOpening()
       this.openState = 'error'
       this.openError = openFailure(error)
     } finally {
@@ -623,10 +577,10 @@ export class Session implements SessionFace {
   }
 
   /** Apply one contiguous journal update already reconciled by the Remote stream. */
-  private acceptEventChange(change: SessionJournalChange, generation: number): void {
+  private acceptEventChange(change: SessionJournalChange): void {
     switch (change.type) {
       case 'replace':
-        this.installWindow(change.entries, change.hasMore, generation, change.page.projections)
+        this.installWindow(change.entries, change.hasMore, change.page.projections)
         return
       case 'prepend':
         this.prependWindow(change.entries, change.hasMore)
@@ -640,23 +594,19 @@ export class Session implements SessionFace {
   private installWindow(
     entries: readonly SessionEventLikeEntry[],
     hasMore: boolean,
-    generation: number,
     projections?: ProjectionsBaseline,
   ): void {
     this.baseSeq = entries[0]?.event.seq ?? 0
     this.hasMore = hasMore
     if (entries.some(entry => entry.event.type === 'turn/start')) this.firstPromptPendingTurn = false
-    const capture = this.projectionCapture?.generation === generation
-      ? this.projectionCapture
-      : undefined
-    if (projections !== undefined && capture !== undefined) {
-      this.projections.replace(projections)
-      this.exactProjectionBaselineInstalled = true
-      this.projectionCapture = undefined
-      this.replayProjectionCapture(projections, capture)
+    if (projections !== undefined) {
+      const opening = this.projectionOpening
+      /* v8 ignore next -- only follow snapshots carry projections, and every follow generation starts a token. */
+      if (opening === undefined) throw new Error('projection baseline arrived outside an opening')
+      this.projections.completeOpening(opening, projections)
+      this.projectionOpening = undefined
     } else {
-      if (projections !== undefined) this.projections.seed(projections)
-      if (capture !== undefined) this.projectionCapture = undefined
+      this.cancelProjectionOpening()
     }
     this.eventSource.replace(entries, hasMore)
     this.notifier.markDirty()
@@ -682,7 +632,7 @@ export class Session implements SessionFace {
   /** Publish a terminal background failure only while this stream still owns the Session. */
   private failEventStream(events: SessionEventStream, generation: number, error: unknown): void {
     if (generation !== this.openGeneration || this.events !== events) return
-    this.discardProjectionCapture(generation)
+    this.cancelProjectionOpening()
     this.openGeneration++
     this.events = undefined
     this.openPromise = null
@@ -692,48 +642,16 @@ export class Session implements SessionFace {
     this.notifier.markDirty()
   }
 
-  /** Start one operation-local capture without dropping operations from a repeated carrier failure. */
-  private beginProjectionCapture(generation: number): void {
-    if (this.projectionCapture?.generation === generation) return
-    this.projectionCapture = { generation, frames: [] }
-  }
-
-  /** Retain one frame only while this generation awaits its exact baseline. */
-  private captureProjectionFrame(frame: ProjectionFrame): void {
-    this.projectionCapture?.frames.push(frame)
-  }
-
-  /** A replacement baseline supersedes every earlier captured control operation. */
-  private captureProjectionBaseline(baseline: ProjectionsBaseline): void {
-    const capture = this.projectionCapture
-    if (capture === undefined) return
-    capture.baseline = baseline
-    capture.frames.length = 0
-  }
-
-  /** Merge normalized control input over one exact opening cut without regressing it. */
-  private replayProjectionCapture(opening: ProjectionsBaseline, capture: ProjectionCapture): void {
-    const baseline = capture.baseline
-    let replayCut = opening.asOfSeq
-    if (baseline !== undefined && baseline.asOfSeq >= replayCut) {
-      this.applyProjectionBaseline(baseline)
-      replayCut = baseline.asOfSeq
-    }
-    for (const frame of capture.frames) {
-      if (frame.seq <= replayCut) continue
-      this.projections.apply(frame.key, frame.value, frame.seq)
-    }
-  }
-
-  /** Apply one complete control-stream replacement to the live store. */
-  private applyProjectionBaseline(baseline: ProjectionsBaseline): void {
-    this.projections.truncate(baseline.asOfSeq)
-    this.projections.seed(baseline)
+  /** Start one store-owned reconciliation interval, retaining it across repeated carrier failures. */
+  private beginProjectionOpening(): void {
+    this.projectionOpening ??= this.projections.beginOpening()
   }
 
-  /** Drop only the capture owned by a failed or superseded generation. */
-  private discardProjectionCapture(generation: number): void {
-    if (this.projectionCapture?.generation === generation) this.projectionCapture = undefined
+  /** End the current reconciliation interval without altering visible values. */
+  private cancelProjectionOpening(): void {
+    const opening = this.projectionOpening
+    this.projectionOpening = undefined
+    if (opening !== undefined) this.projections.cancelOpening(opening)
   }
 
   private buildSnapshot(): SessionSnapshot {

packages/api/session-controller/tests/manager.client.spec.ts

@@ -151,11 +151,11 @@ describe('list lifecycle', () => {
     expect(manager.getListSnapshot().items.find(item => item.sessionId === S1)?.title).toBeUndefined()
   })
 
-  it('applies cold title values from list and session-added blocks by sequence', async () => {
+  it('prewarms cold titles without letting later hints override an authoritative frame', async () => {
     const api = new FakeApiClient()
     const manager = new SessionManager(fakeRemote(api))
-    // A push frame landed before the list. A later cached cut wins under the
-    // same sequence rule used by every projection source.
+    // A push frame landed before the list. Its authoritative value wins even
+    // when a later cache hint claims a higher sequence.
     manager.handleControlFrame({
       type: 'projection', sessionId: S2, key: 'title', value: 'Pushed', seq: 9,
     })
@@ -169,21 +169,23 @@ describe('list lifecycle', () => {
     const items = manager.getListSnapshot().items
     // Cold row: title surfaces straight from the list block — no open, no history.
     expect(items.find(item => item.sessionId === S1)?.title).toBe('Cold cached')
-    expect(items.find(item => item.sessionId === S2)?.title).toBe('List stale')
+    expect(items.find(item => item.sessionId === S2)?.title).toBe('Pushed')
     manager.handleSessionAdded({
       ...summary(S2, { updatedAt: 300 }),
       projections: { asOfSeq: 15, values: { title: 'Added stale' } },
     })
-    expect(manager.getListSnapshot().items.find(item => item.sessionId === S2)?.title).toBe('Added stale')
+    expect(manager.getListSnapshot().items.find(item => item.sessionId === S2)?.title).toBe('Pushed')
   })
 
-  it('drops a projection row beyond the subscription baseline before accepting its durable replay', async () => {
+  it('routes resident and list projections through one store across exact control replacements', async () => {
     const api = new FakeApiClient()
     api.onList = () => Promise.resolve(ok({ items: [summary(S1)] as never[] }))
     const manager = new SessionManager(fakeRemote(api))
     await manager.refreshList()
+    const session = manager.get(S1)
     const frame = (payload: SessionControlFrame) => { manager.handleControlFrame(payload) }
     frame({ type: 'projection', sessionId: S1, key: 'title', value: 'Unflushed', seq: 4 })
+    expect(session.projections.get('title')).toBe('Unflushed')
 
     // The durable baseline says the host only knows up to seq 2: the phantom
     // row rode lost state and must drop, or last-wins pins it forever.
@@ -195,19 +197,21 @@ describe('list lifecycle', () => {
       },
     })
     expect(manager.getListSnapshot().items[0]?.title).toBeUndefined()
+    expect(session.projections.get('title')).toBeUndefined()
 
     frame({ type: 'projection', sessionId: S1, key: 'title', value: 'Durable', seq: 2 })
     expect(manager.getListSnapshot().items[0]?.title).toBe('Durable')
 
-    // A baseline at or past the row's seq keeps it (nothing phantom to drop).
+    // The next complete generation replaces even a different equal-sequence row.
     frame({
       type: 'baseline',
       value: {
         queues: {}, jobs: {},
-        projections: { [S1]: { asOfSeq: 2, values: { title: 'Durable' } } },
+        projections: { [S1]: { asOfSeq: 2, values: { title: 'Recovered' } } },
       },
     })
-    expect(manager.getListSnapshot().items[0]?.title).toBe('Durable')
+    expect(manager.getListSnapshot().items[0]?.title).toBe('Recovered')
+    expect(session.projections.get('title')).toBe('Recovered')
   })
 })
 

packages/api/session-controller/tests/projection-store.client.spec.ts

@@ -1,21 +1,15 @@
 /**
- * Projection value store (push model; session-projection subsystem page:
- * docs/subsystems/session-projection.md): higher-seq-wins for ordinary inputs,
- * exact opening replacement, capability absence as undefined, generation truncation, and the
- * Session/manager wiring (tail-page seeding, control-stream projection routing
- * pre- and post-instantiation, and list-row projection values).
+ * Projection value-store precedence plus the minimal Session opening lifecycle
+ * that creates and settles store-owned reconciliation tokens.
  */
 import { describe, expect, it, vi } from 'vitest'
 import { RemoteStreamCarrierError } from '@deepseek-ai/dsh-api-gateway/client'
 import type { SessionId } from '@deepseek-ai/dsh-api-remotes/client'
 import { ProjectionValueStore } from '../src/client/sessions/projection-store.ts'
 import { Session } from '../src/client/sessions/session.ts'
-import { SessionManager } from '../src/client/sessions/manager.ts'
 import { FakeApiClient, deferred, err, fakeRemote, ok } from './fake-api.client.ts'
 import { entries, plainTurn } from './event-script.client.ts'
 
-// Test-domain keys merged into the projection map (the Service Definition package's
-// pure-type outlet), the same way domain host plugins merge theirs.
 declare module '@deepseek-ai/dsh-session-projection/types' {
   interface SessionProjectionMap {
     'test/marks': { marks: string[] }
@@ -25,87 +19,165 @@ declare module '@deepseek-ai/dsh-session-projection/types' {
 const SID = 'fk-s1' as SessionId
 
 describe('Session projection value semantics', () => {
-  it('reads undefined until a value lands (capability absence)', () => {
+  it('reads undefined until a value lands and keeps stable observable faces', () => {
     const store = new ProjectionValueStore()
     expect(store.get('test/marks')).toBeUndefined()
     expect(store.faceOf('test/marks').getSnapshot()).toBeUndefined()
+    expect(store.faceOf('test/marks')).toBe(store.faceOf('test/marks'))
   })
 
-  it('applies frames last-wins by seq: replayed and stale frames drop', () => {
+  it('lets the first authoritative frame replace a newer hint, then uses higher-seq-wins', () => {
     const store = new ProjectionValueStore()
-    store.apply('test/marks', { marks: ['a'] }, 5)
-    store.apply('test/marks', { marks: ['a', 'b'] }, 9)
-    expect(store.get('test/marks')).toEqual({ marks: ['a', 'b'] })
-    store.apply('test/marks', { marks: ['stale'] }, 5)
-    store.apply('test/marks', { marks: ['equal'] }, 9)
-    expect(store.get('test/marks')).toEqual({ marks: ['a', 'b'] })
+    store.prewarm({
+      asOfSeq: 9,
+      values: { 'test/marks': { marks: ['hint-9'] }, 'hint-only': 'hint' },
+    })
+    store.prewarm({
+      asOfSeq: 8,
+      values: { 'test/marks': { marks: ['older-hint'] } },
+    })
+
+    store.apply('test/marks', { marks: ['frame-3'] }, 3)
+    store.prewarm({
+      asOfSeq: 12,
+      values: { 'test/marks': { marks: ['late-hint'] }, 'other-hint': 'other' },
+    })
+    store.apply('test/marks', { marks: ['equal-frame'] }, 3)
+    store.apply('test/marks', { marks: ['newer-frame'] }, 4)
+
+    expect(store.values()).toEqual({
+      'test/marks': { marks: ['newer-frame'] },
+      'hint-only': 'hint',
+      'other-hint': 'other',
+    })
   })
 
-  it('uses the same higher-seq-wins rule for cached and live values', () => {
+  it('opening replaces pre-opening state and retains only newer post-token frames', () => {
     const store = new ProjectionValueStore()
-    store.apply('test/marks', { marks: ['cached-5'] }, 5)
-    store.apply('test/marks', { marks: ['stale-live'] }, 3)
-    store.apply('test/marks', { marks: ['cached-9'] }, 9)
-    store.apply('test/marks', { marks: ['equal-live'] }, 9)
-    expect(store.get('test/marks')).toEqual({ marks: ['cached-9'] })
+    store.prewarm({ asOfSeq: 20, values: { 'test/marks': { marks: ['hint'] } } })
+    store.apply('pre-opening', 'old-frame', 30)
+    const opening = store.beginOpening()
+    store.apply('test/marks', { marks: ['post-token'] }, 3)
+    store.apply('discarded', 'at-cut', 10)
+    store.apply('retained', 'new-frame', 11)
+
+    store.completeOpening(opening, {
+      asOfSeq: 10,
+      values: {
+        'test/marks': { marks: ['opening'] },
+        'opening-only': 'present',
+      },
+    })
+
+    expect(store.values()).toEqual({
+      'test/marks': { marks: ['opening'] },
+      'opening-only': 'present',
+      retained: 'new-frame',
+    })
   })
 
-  it('a complete baseline updates and clears only rows at or below its cut', () => {
+  it('a control baseline exactly replaces equal-sequence and omitted rows', () => {
     const store = new ProjectionValueStore()
-    store.apply('test/marks', { marks: ['old'] }, 5)
-    store.apply('cleared', 'old', 5)
-    store.apply('newer', 'newer', 20)
-    store.seed({ asOfSeq: 10, values: { 'test/marks': { marks: ['baseline-10'] } } })
-    expect(store.get('test/marks')).toEqual({ marks: ['baseline-10'] })
-    expect(store.get('cleared')).toBeUndefined()
-    expect(store.get('newer')).toBe('newer')
-    store.seed({ asOfSeq: 10, values: {} })
-    expect(store.get('test/marks')).toBeUndefined()
-    expect(store.get('newer')).toBe('newer')
+    store.apply('title', 'transient', 1)
+    store.apply('omitted', 'transient', 1)
+
+    store.replaceControlBaseline({ asOfSeq: 1, values: { title: null } })
+
+    expect(store.values()).toEqual({ title: null })
+    store.apply('title', 'equal-frame', 1)
+    expect(store.get('title')).toBeNull()
+    store.apply('title', 'durable-frame', 2)
+    expect(store.get('title')).toBe('durable-frame')
+  })
+
+  it('an equal or newer control baseline received during opening wins', () => {
+    const store = new ProjectionValueStore()
+    const opening = store.beginOpening()
+    store.replaceControlBaseline({
+      asOfSeq: 5,
+      values: { 'test/marks': { marks: ['control'] }, 'control-only': 'present' },
+    })
+    store.apply('test/marks', { marks: ['after-control'] }, 6)
+
+    store.completeOpening(opening, {
+      asOfSeq: 5,
+      values: { 'test/marks': { marks: ['opening'] }, 'opening-only': 'discarded' },
+    })
+
+    expect(store.values()).toEqual({
+      'test/marks': { marks: ['after-control'] },
+      'control-only': 'present',
+    })
   })
 
-  it('an exact baseline replaces every prior row even when its cut is lower', () => {
+  it('a newer opening replaces an older control baseline but keeps later frames', () => {
     const store = new ProjectionValueStore()
-    store.apply('test/marks', { marks: ['ghost'] }, 9)
-    store.apply('omitted', 'ghost', 9)
-    store.replace({ asOfSeq: 2, values: { 'test/marks': { marks: ['durable'] } } })
-    expect(store.get('test/marks')).toEqual({ marks: ['durable'] })
-    expect(store.get('omitted')).toBeUndefined()
-    store.apply('test/marks', { marks: ['live'] }, 3)
-    expect(store.get('test/marks')).toEqual({ marks: ['live'] })
+    const opening = store.beginOpening()
+    store.replaceControlBaseline({
+      asOfSeq: 5,
+      values: { 'test/marks': { marks: ['control'] }, 'control-only': 'discarded' },
+    })
+    store.apply('stale-frame', 'discarded', 6)
+    store.apply('retained-frame', 'present', 11)
+
+    store.completeOpening(opening, {
+      asOfSeq: 10,
+      values: { 'test/marks': { marks: ['opening'] }, 'opening-only': 'present' },
+    })
+
+    expect(store.values()).toEqual({
+      'test/marks': { marks: ['opening'] },
+      'opening-only': 'present',
+      'retained-frame': 'present',
+    })
   })
 
-  it('truncate drops rows past the durable baseline and keeps the rest', () => {
+  it('ignores hints after a complete baseline and rejects stale opening tokens', () => {
     const store = new ProjectionValueStore()
-    store.apply('test/marks', { marks: ['durable'] }, 5)
-    store.apply('other', 'phantom', 50)
-    store.truncate(10)
-    expect(store.get('test/marks')).toEqual({ marks: ['durable'] })
-    expect(store.get('other')).toBeUndefined()
+    const stale = store.beginOpening()
+    const current = store.beginOpening()
+    store.completeOpening(stale, { asOfSeq: 1, values: { stale: true } })
+    expect(store.values()).toEqual({})
+    store.cancelOpening(stale)
+    store.completeOpening(current, { asOfSeq: 1, values: { exact: true } })
+    store.prewarm({ asOfSeq: 99, values: { exact: false, late: true } })
+    expect(store.values()).toEqual({ exact: true })
   })
 
-  it('notifies the key face on change (batched) and not on dropped applications', async () => {
+  it('a canceled opening makes its frames pre-opening for the next exact cut', () => {
+    const store = new ProjectionValueStore()
+    const failed = store.beginOpening()
+    store.apply('test/marks', { marks: ['failed-frame'] }, 3)
+    store.cancelOpening(failed)
+    const retry = store.beginOpening()
+    store.completeOpening(retry, {
+      asOfSeq: 2,
+      values: { 'test/marks': { marks: ['retry'] } },
+    })
+    expect(store.get('test/marks')).toEqual({ marks: ['retry'] })
+  })
+
+  it('notifies accepted changes but not dropped authoritative frames', async () => {
     const store = new ProjectionValueStore()
     let keyTicks = 0
     let anyTicks = 0
     store.faceOf('test/marks').subscribe(() => { keyTicks += 1 })
     store.subscribeAny(() => { anyTicks += 1 })
-    store.apply('test/marks', { marks: ['a'] }, 5)
+    const value = { marks: ['a'] }
+    store.apply('test/marks', value, 1)
     await Promise.resolve()
     expect(keyTicks).toBe(1)
     expect(anyTicks).toBe(1)
-    store.apply('test/marks', { marks: ['replay'] }, 3)
+    store.apply('test/marks', { marks: ['replay'] }, 0)
+    store.replaceControlBaseline({ asOfSeq: 5, values: { 'test/marks': value } })
+    store.apply('test/marks', { marks: ['stale-after-baseline'] }, 4)
     await Promise.resolve()
     expect(keyTicks).toBe(1)
     expect(anyTicks).toBe(1)
+    expect(store.get('test/marks')).toBe(value)
   })
 
-  it('faces are identity-stable per key (the React binding cache premise)', () => {
-    const store = new ProjectionValueStore()
-    expect(store.faceOf('test/marks')).toBe(store.faceOf('test/marks'))
-  })
-
-  it('publishes one reference-stable whole-value snapshot until a row changes', () => {
+  it('publishes one stable whole-value snapshot until an observable row changes', () => {
     const store = new ProjectionValueStore()
     const empty = store.values()
     expect(store.values()).toBe(empty)
@@ -117,11 +189,11 @@ describe('Session projection value semantics', () => {
   })
 })
 
-describe('Session tail-page seeding', () => {
-  it('retains a prewarmed projection when opening the Session fails', async () => {
+describe('Session opening integration', () => {
+  it('retains a tentative hint when opening fails, then replaces it on retry', async () => {
     const api = new FakeApiClient()
     const projections = new ProjectionValueStore()
-    projections.apply('test/marks', { marks: ['cached'] }, 5)
+    projections.prewarm({ asOfSeq: 5, values: { 'test/marks': { marks: ['cached'] } } })
     const session = new Session(SID, fakeRemote(api), { projections })
     api.onHistory = () => Promise.resolve(err({
       code: 'session-not-found',
@@ -130,215 +202,35 @@ describe('Session tail-page seeding', () => {
     }))
 
     await session.open()
-
     expect(session.getSnapshot().openState).toBe('error')
     expect(session.projections.get('test/marks')).toEqual({ marks: ['cached'] })
-  })
 
-  it('seeds the store from a history response carrying a projections block', async () => {
-    const api = new FakeApiClient()
-    const session = new Session(SID, fakeRemote(api))
-    api.onHistory = () => Promise.resolve(ok({
-      records: entries(plainTurn(0, 0, '问', '答')) as never[], hasMore: false,
-      projections: { asOfSeq: 5, values: { 'test/marks': { marks: ['from-baseline'] } } },
-    } as never))
-    await session.open()
-    expect(session.projections.get('test/marks')).toEqual({ marks: ['from-baseline'] })
-  })
-
-  it('replaces a higher-sequence cache ghost with the exact opening baseline', async () => {
-    const api = new FakeApiClient()
-    const projections = new ProjectionValueStore()
-    projections.apply('test/marks', { marks: ['cached'] }, 9)
-    const session = new Session(SID, fakeRemote(api), { projections })
     api.onHistory = () => Promise.resolve(ok({
       records: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false,
-      projections: { asOfSeq: 2, values: { 'test/marks': { marks: ['older-baseline'] } } },
-    } as never))
-
-    await session.open()
-
-    expect(session.getSnapshot().openState).toBe('open')
-    expect(session.projections.get('test/marks')).toEqual({ marks: ['older-baseline'] })
-  })
-
-  it('replays live control frames after replacing cache hints during opening', async () => {
-    const api = new FakeApiClient()
-    const history = deferred<Awaited<ReturnType<FakeApiClient['onHistory']>>>()
-    api.onHistory = () => history.promise
-    const projections = new ProjectionValueStore()
-    projections.apply('test/marks', { marks: ['cached-9'] }, 9)
-    const session = new Session(SID, fakeRemote(api), { projections })
-
-    const opening = session.open()
-    session.handleProjectionFrame({
-      type: 'projection', sessionId: SID, key: 'test/marks', value: { marks: ['live-3'] }, seq: 3,
-    })
-    history.resolve(ok({
-      records: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false,
-      projections: { asOfSeq: 2, values: { 'test/marks': { marks: ['baseline-2'] } } },
-    } as never))
-    await opening
-
-    expect(session.projections.get('test/marks')).toEqual({ marks: ['live-3'] })
-  })
-
-  it('resync removes pre-operation high rows and replays only control operations that arrive during resync', async () => {
-    const api = new FakeApiClient()
-    const session = new Session(SID, fakeRemote(api))
-    api.onHistory = () => Promise.resolve(ok({
-      records: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false,
-      projections: { asOfSeq: 5, values: { 'test/marks': { marks: ['baseline'] } } },
+      projections: { asOfSeq: 2, values: { 'test/marks': { marks: ['exact'] } } },
     } as never))
     await session.open()
-    session.handleProjectionFrame({
-      type: 'projection', sessionId: SID, key: 'test/marks', value: { marks: ['old-9'] }, seq: 9,
-    })
-    const history = deferred<Awaited<ReturnType<FakeApiClient['onHistory']>>>()
-    api.onHistory = () => history.promise
-    const resyncing = session.resync()
-    session.handleProjectionFrame({
-      type: 'projection', sessionId: SID, key: 'test/marks', value: { marks: ['during-3'] }, seq: 3,
-    })
-    history.resolve(ok({
-      records: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false,
-      projections: { asOfSeq: 2, values: { 'test/marks': { marks: ['baseline-2'] } } },
-    } as never))
-    await resyncing
-    expect(session.projections.get('test/marks')).toEqual({ marks: ['during-3'] })
-  })
-
-  it('replays a newer control baseline and only its subsequent frames', async () => {
-    const api = new FakeApiClient()
-    const session = new Session(SID, fakeRemote(api))
-    const history = deferred<Awaited<ReturnType<FakeApiClient['onHistory']>>>()
-    api.onHistory = () => history.promise
-    const opening = session.open()
-    session.handleProjectionFrame({
-      type: 'projection', sessionId: SID, key: 'test/marks', value: { marks: ['first-frame'] }, seq: 7,
-    })
-    session.replaceProjectionBaseline({
-      asOfSeq: 2, values: { 'test/marks': { marks: ['control-baseline'] } },
-    })
-    session.handleProjectionFrame({
-      type: 'projection', sessionId: SID, key: 'test/marks', value: { marks: ['last-frame'] }, seq: 3,
-    })
-    history.resolve(ok({
-      records: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false,
-      projections: { asOfSeq: 1, values: { 'test/marks': { marks: ['opening'] } } },
-    } as never))
-    await opening
-    expect(session.projections.get('test/marks')).toEqual({ marks: ['last-frame'] })
+    expect(session.projections.get('test/marks')).toEqual({ marks: ['exact'] })
   })
 
-  it('keeps a newer exact opening over an older captured control baseline', async () => {
+  it('preserves an authoritative frame received while the opening is in flight', async () => {
     const api = new FakeApiClient()
-    const session = new Session(SID, fakeRemote(api))
     const history = deferred<Awaited<ReturnType<FakeApiClient['onHistory']>>>()
     api.onHistory = () => history.promise
-
-    const opening = session.open()
-    session.replaceProjectionBaseline({
-      asOfSeq: 5, values: { 'test/marks': { marks: ['control-5'] } },
-    })
-    session.handleProjectionFrame({
-      type: 'projection', sessionId: SID, key: 'stale', value: 'frame-6', seq: 6,
-    })
-    history.resolve(ok({
-      records: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false,
-      projections: {
-        asOfSeq: 10,
-        values: {
-          'test/marks': { marks: ['opening-10'] },
-          'opening-only': 'present',
-        },
-      },
-    } as never))
-    await opening
-
-    expect(session.projections.values()).toEqual({
-      'test/marks': { marks: ['opening-10'] },
-      'opening-only': 'present',
-    })
-  })
-
-  it('uses the latest captured baseline generation before merging later frames', async () => {
-    const api = new FakeApiClient()
     const session = new Session(SID, fakeRemote(api))
-    const history = deferred<Awaited<ReturnType<FakeApiClient['onHistory']>>>()
-    api.onHistory = () => history.promise
 
     const opening = session.open()
-    session.replaceProjectionBaseline({
-      asOfSeq: 12, values: { 'test/marks': { marks: ['superseded-control-12'] } },
-    })
-    session.handleProjectionFrame({
-      type: 'projection', sessionId: SID, key: 'superseded', value: 'frame-13', seq: 13,
-    })
-    session.replaceProjectionBaseline({
-      asOfSeq: 8, values: { 'test/marks': { marks: ['latest-control-8'] } },
-    })
-    session.handleProjectionFrame({
-      type: 'projection', sessionId: SID, key: 'test/marks', value: { marks: ['frame-11'] }, seq: 11,
-    })
+    session.projections.apply('test/marks', { marks: ['live'] }, 3)
     history.resolve(ok({
       records: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false,
-      projections: {
-        asOfSeq: 10,
-        values: {
-          'test/marks': { marks: ['opening-10'] },
-          'opening-only': 'present',
-        },
-      },
+      projections: { asOfSeq: 2, values: { 'test/marks': { marks: ['opening'] } } },
     } as never))
     await opening
 
-    expect(session.projections.values()).toEqual({
-      'test/marks': { marks: ['frame-11'] },
-      'opening-only': 'present',
-    })
-  })
-
-  it('ignores a list hint after the exact baseline is installed but before open settles', async () => {
-    const api = new FakeApiClient()
-    const session = new Session(SID, fakeRemote(api))
-    api.onHistory = () => Promise.resolve(ok({
-      records: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false,
-      projections: { asOfSeq: 2, values: { 'test/marks': { marks: ['exact'] } } },
-    } as never))
-    const unsubscribe = session.eventSource.subscribe(() => {
-      expect(session.getSnapshot().openState).toBe('loading')
-      session.handleProjectionHint({
-        asOfSeq: 99, values: { 'test/marks': { marks: ['late-hint'] } },
-      })
-    })
-    await session.open()
-    unsubscribe()
-    expect(session.projections.get('test/marks')).toEqual({ marks: ['exact'] })
-  })
-
-  it('keeps normally applied control state on failure without replaying the failed capture into a later open', async () => {
-    const api = new FakeApiClient()
-    const first = deferred<Awaited<ReturnType<FakeApiClient['onHistory']>>>()
-    api.onHistory = () => first.promise
-    const session = new Session(SID, fakeRemote(api))
-    const failedOpen = session.open()
-    session.handleProjectionFrame({
-      type: 'projection', sessionId: SID, key: 'test/marks', value: { marks: ['during-failure'] }, seq: 3,
-    })
-    first.resolve(err({ code: 'session-not-found', message: 'gone', details: { sessionId: SID } }))
-    await failedOpen
-    expect(session.projections.get('test/marks')).toEqual({ marks: ['during-failure'] })
-
-    api.onHistory = () => Promise.resolve(ok({
-      records: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false,
-      projections: { asOfSeq: 2, values: { 'test/marks': { marks: ['later-open'] } } },
-    } as never))
-    await session.open()
-    expect(session.projections.get('test/marks')).toEqual({ marks: ['later-open'] })
+    expect(session.projections.get('test/marks')).toEqual({ marks: ['live'] })
   })
 
-  it('replays frames received while a carrier reconnect waits for its replacement snapshot', async () => {
+  it('starts a fresh opening token while a carrier reconnect awaits its snapshot', async () => {
     const api = new FakeApiClient()
     const session = new Session(SID, fakeRemote(api))
     api.onHistory = () => Promise.resolve(ok({
@@ -351,126 +243,14 @@ describe('Session tail-page seeding', () => {
     api.onHistory = () => replacement.promise
     api.failStreams(new RemoteStreamCarrierError('carrier lost'))
     await vi.waitFor(() => { expect(api.callsOf('session.follow')).toHaveLength(2) })
-    session.handleProjectionFrame({
-      type: 'projection', sessionId: SID, key: 'test/marks', value: { marks: ['during-retry'] }, seq: 3,
-    })
+    session.projections.apply('test/marks', { marks: ['during-retry'] }, 3)
     replacement.resolve(ok({
       records: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false,
       projections: { asOfSeq: 2, values: { 'test/marks': { marks: ['replacement'] } } },
     } as never))
+
     await vi.waitFor(() => {
       expect(session.projections.get('test/marks')).toEqual({ marks: ['during-retry'] })
     })
   })
 })
-
-describe('manager frame routing', () => {
-  const sid = (s: string): SessionId => s as SessionId
-
-  it('lands projection frames before instantiation and the Session adopts the same store', async () => {
-    const api = new FakeApiClient()
-    const manager = new SessionManager(fakeRemote(api))
-    manager.handleControlFrame({
-      type: 'projection', sessionId: sid('s1'), key: 'test/marks', value: { marks: ['early'] }, seq: 7,
-    })
-    const session = manager.get(sid('s1'))
-    expect(session.projections.get('test/marks')).toEqual({ marks: ['early'] })
-    // Frames after instantiation land in the same store.
-    manager.handleControlFrame({
-      type: 'projection', sessionId: sid('s1'), key: 'test/marks', value: { marks: ['later'] }, seq: 9,
-    })
-    expect(session.projections.get('test/marks')).toEqual({ marks: ['later'] })
-  })
-
-  it('projects the title key into list rows and truncates phantom rows on the control baseline', async () => {
-    const api = new FakeApiClient()
-    const manager = new SessionManager(fakeRemote(api))
-    api.onList = () => Promise.resolve(ok({
-      items: [{ sessionId: sid('s1'), updatedAt: 1, running: false, blank: false }],
-    }) as never)
-    await manager.refreshList()
-    manager.get(sid('s1'))
-    manager.handleControlFrame({
-      type: 'projection', sessionId: sid('s1'), key: 'title', value: 'Projected title', seq: 4,
-    })
-    await Promise.resolve()
-    expect(manager.getListSnapshot().items[0]?.title).toBe('Projected title')
-    // The durable baseline says the host only knows up to seq 2: the row rode
-    // lost state and must drop (the un-flushed title precedent).
-    manager.handleControlFrame({
-      type: 'baseline',
-      value: {
-        queues: {}, jobs: {},
-        projections: { [sid('s1')]: { asOfSeq: 2, values: {} } },
-      },
-    })
-    await Promise.resolve()
-    expect(manager.getListSnapshot().items[0]?.title).toBeUndefined()
-  })
-
-  it('projects every retained value into list rows with stable snapshot identity', async () => {
-    const api = new FakeApiClient()
-    const manager = new SessionManager(fakeRemote(api))
-    api.onList = () => Promise.resolve(ok({
-      items: [{
-        sessionId: sid('s1'), updatedAt: 1, running: false, blank: false,
-        projections: {
-          asOfSeq: 2,
-          values: { 'test/marks': { marks: ['baseline'] } },
-        },
-      }],
-    }) as never)
-    await manager.refreshList()
-    const baseline = manager.getListSnapshot().items[0]?.projectionValues
-    expect(baseline).toEqual({ 'test/marks': { marks: ['baseline'] } })
-    expect(manager.getListSnapshot().items[0]?.projectionValues).toBe(baseline)
-
-    manager.handleControlFrame({
-      type: 'projection', sessionId: sid('s1'), key: 'test/marks',
-      value: { marks: ['live'] }, seq: 3,
-    })
-    await Promise.resolve()
-    expect(manager.getListSnapshot().items[0]?.projectionValues)
-      .toEqual({ 'test/marks': { marks: ['live'] } })
-    expect(manager.getListSnapshot().items[0]?.projectionValues).not.toBe(baseline)
-  })
-
-  it('keeps late list and session-added cache hints out of an already opened Session', async () => {
-    const api = new FakeApiClient()
-    const manager = new SessionManager(fakeRemote(api))
-    const sessionId = sid('s1')
-    api.onHistory = () => Promise.resolve(ok({
-      records: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false,
-      projections: { asOfSeq: 2, values: { 'test/marks': { marks: ['exact'] } } },
-    } as never))
-    const session = manager.get(sessionId)
-    await session.open()
-
-    api.onList = () => Promise.resolve(ok({
-      items: [{
-        sessionId, updatedAt: 1, running: false, blank: false,
-        projections: { asOfSeq: 99, values: { 'test/marks': { marks: ['list-hint'] } } },
-      }],
-    }) as never)
-    await manager.refreshList()
-    manager.handleSessionAdded({
-      sessionId, updatedAt: 2, running: false, blank: false,
-      projections: { asOfSeq: 100, values: { 'test/marks': { marks: ['added-hint'] } } },
-    })
-    expect(session.projections.get('test/marks')).toEqual({ marks: ['exact'] })
-  })
-
-  it('drops the projection store with the removed session', async () => {
-    const api = new FakeApiClient()
-    const manager = new SessionManager(fakeRemote(api))
-    api.onList = () => Promise.resolve(ok({
-      items: [{ sessionId: sid('s1'), updatedAt: 1, running: false, blank: false }],
-    }) as never)
-    await manager.refreshList()
-    manager.handleControlFrame({
-      type: 'projection', sessionId: sid('s1'), key: 'title', value: 'Doomed', seq: 4,
-    })
-    manager.handleSessionRemoved(sid('s1'))
-    expect(manager.get(sid('s1')).projections.get('title')).toBeUndefined()
-  })
-})

packages/session/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/session-projection-cache/README.md
-README.md: 03db887ae481b554e38d7599321bc727aa554a52
-README.zh.md: f4abde11ed81e620c16992f4eecd8f2032346b5a
+README.md: 9d1619fe25075577218e95a843f752ce3cec7cfb
+README.zh.md: cd72f124865df32b8576d7f6dd3aff12a3a1cd05

packages/session/session-projection-cache/README.md

@@ -58,7 +58,7 @@ Three mandatory points always write: session creation persists the seed-derived
 
 ### Reading cached values
 
-`cachedSnapshot(meta)` synchronously serves client values from the storage domain's coherent in-memory table with zero I/O. It accepts only an identity-matching record and version- and schema-matching client keys, then returns a best-effort `{ asOfSeq, values }` cut at the lowest served-row watermark; host-only rows are omitted. It returns `undefined` for an unknown id, unrelated lifecycle, absent or foreign record document, or no usable rows. The list carrier uses this value only as a tentative hint: a successful follow opening replaces it exactly, then replays control updates that arrived during the opening. Ordinary hints and live frames remain higher-sequence-wins, and replacement control baselines may truncate rows beyond their durable cut.
+`cachedSnapshot(meta)` synchronously serves client values from the storage domain's coherent in-memory table with zero I/O. It accepts only an identity-matching record and version- and schema-matching client keys, then returns a best-effort `{ asOfSeq, values }` cut at the lowest served-row watermark; host-only rows are omitted. It returns `undefined` for an unknown id, unrelated lifecycle, absent or foreign record document, or no usable rows. The list carrier submits this value as a tentative hint to the per-Session Client projection store. Higher-sequence hints replace earlier tentative rows; the first authoritative frame replaces a tentative row regardless of sequence, and later frames require a higher sequence. A successful follow opening gives the store a complete cut: it replaces pre-opening rows and retains only authoritative frames that arrived after the opening began and are newer than that cut. A control-generation baseline also replaces its complete per-Session value, including equal-sequence and omitted keys; when it arrives during an opening at an equal or newer cut, it remains authoritative.
 
 `coldSnapshot(meta, events)` accepts the complete ordered log, validates every seeded row against that exact extent once, folds any required events from `init(header)`, and refreshes the record without consulting persistence. `hydratePrepared(session, meta, events)` performs the production exact-read validation for an unpublished prepared Session; if cached state is malformed or out of range, that path retries over the full supplied log from `init(header)`. Corruption in the durable event stream still fails the retry instead of producing a partial snapshot.
 
@@ -127,7 +127,7 @@ These limits define where the cache needs operational care. They are current pac
 
 - **No eviction or retention surface** — records accumulate per session; pruning stored checkpoints is out-of-band maintenance, same stance as session persistence itself.
 - **Interval throttle is per-session coarse** — the timer arms at the first dirty event after a clean write; a steady sub-threshold trickle writes once per interval, not a sliding window.
-- **Zero-I/O values are best effort** — a cached row may trail current events or overreach a crash-repaired truncation; exact Host reads validate against the complete log, while the client keeps the highest sequence until a later value or replacement control baseline supersedes it.
+- **Zero-I/O values are best effort** — a cached row may trail current events or overreach a crash-repaired truncation; exact Host reads validate against the complete log, while the Client treats each list value as tentative until the first authoritative frame or complete follow/control baseline replaces it under the projection store's precedence rules.
 - **Callers supply cold logs** — the cache can validate and refold a complete log but never reads session persistence itself; a consumer that needs an exact cold snapshot owns that log read.
 
 <a id="dev-note"></a>

packages/session/session-projection-cache/README.zh.md

@@ -58,7 +58,7 @@ kind: "package-reference"
 
 ### 读取缓存值
 
-`cachedSnapshot(meta)` 以零 I/O 从存储域一致的内存表同步提供客户端值。它只接受身份匹配的记录及版本和 schema 均匹配的客户端 key，再按所服务行的最低水位返回尽力而为的 `{ asOfSeq, values }` 切面；host-only 行会被省略。对于未知 id、无关生命周期、缺失或外来的记录文档，或没有可用行的情况，它返回 `undefined`。列表载体只把该值作为暂存 hint：成功的 follow opening 会精确替换它，再重放 opening 期间到达的 control 更新。普通 hint 与 live frame 继续按 higher-sequence-wins，replacement control baseline 可以截断超出其持久 cut 的行。
+`cachedSnapshot(meta)` 以零 I/O 从存储域一致的内存表同步提供客户端值。它只接受身份匹配的记录及版本和 schema 均匹配的客户端 key，再按所服务行的最低水位返回尽力而为的 `{ asOfSeq, values }` 切面；host-only 行会被省略。对于未知 id、无关生命周期、缺失或外来的记录文档，或没有可用行的情况，它返回 `undefined`。列表载体把该值作为暂定 hint 交给逐 Session 的 Client projection store。较高 sequence 的 hint 会替换较早的暂定 row；首个权威 frame 无论 sequence 如何都会替换暂定 row，后续 frame 则必须具有更高 sequence。成功的 follow opening 为 store 提供一份完整 cut：它替换 opening 前的 row，只保留 opening 开始后到达且新于该 cut 的权威 frame。control generation baseline 也会精确替换该 Session 的完整值，包括等 sequence row 与缺失 key；若它在 opening 期间以等于或新于 opening cut 的 cut 到达，它保持权威。
 
 `coldSnapshot(meta, events)` 接受完整有序日志，只以该精确范围校验一次每条 seed row，从 `init(header)` 折叠所需事件，并在不访问持久化层的情况下刷新记录。`hydratePrepared(session, meta, events)` 为尚未发布的 prepared Session 执行生产精确读取校验；若缓存状态畸形或越界，只有该路径会在所提供的完整日志上从 `init(header)` 重试。持久事件流本身若已损坏，重试仍然失败，绝不会产出部分快照。
 
@@ -127,7 +127,7 @@ kind: "package-reference"
 
 - **无淘汰或保留接口**——记录按会话持续累积；清理已存储检查点属于带外维护，与会话持久化采用相同策略。
 - **间隔节流采用按会话的粗粒度控制**——一次无脏数据的写入完成后，计时器在首个脏事件到达时启动；持续但低于条数阈值的事件流每间隔写入一次，而非滑动窗口。
-- **零 I/O 值是尽力而为的**——缓存行可能落后于当前事件，也可能越过崩溃修复后的截断点；Host 精确读取会用完整日志校验，客户端则保留最高 sequence，直到后续值或 replacement control baseline 取代它。
+- **零 I/O 值是尽力而为的**——缓存行可能落后于当前事件，也可能越过崩溃修复后的截断点；Host 精确读取会用完整日志校验，Client 则把每个 list 值视为暂定值，直到首个权威 frame 或完整 follow/control baseline 按 projection store 的 precedence 规则替换它。
 - **冷日志由调用方提供**——缓存能校验并重新折叠一份完整日志，但绝不自行读取会话持久化层；需要精确冷快照的消费方负责该日志读取。
 
 <a id="dev-note"></a>