diff --git a/packages/api/session-controller/src/client/sessions/manager.ts b/packages/api/session-controller/src/client/sessions/manager.ts index f46b6cc8fe..d9cf53d89f 100644 --- a/packages/api/session-controller/src/client/sessions/manager.ts +++ b/packages/api/session-controller/src/client/sessions/manager.ts @@ -376,18 +376,16 @@ export class SessionManager { session.handleBlank(s.blank) session.handleRunning(s.running) } - // Seed each row's projection baseline into the per-session value - // store (cold titles surface without opening the session). Per-key - // apply, not seed(): the list block is a partial baseline — the - // cold cache serves only version-matching keys — so an absent key - // must not clear; higher-seq-wins still keeps a stale list block - // from overwriting a newer push frame or tail baseline. + // Fill each row's cached projection block into the per-session value + // store (cold titles surface without opening the session). Cached, + // not seeded: the list block is a partial view of the persisted + // checkpoint — only version-matching keys, no comparable seq — so an + // absent key must not clear, and any sequenced value (a push frame, + // a tail baseline) outranks it. for (const s of result.value.items) { const block = s.projections if (block === undefined) continue - const store = this.projectionStore(s.sessionId) - const values = block.values as Record - for (const key of Object.keys(values)) store.apply(key, values[key], sessionSeqCursor(block.asOfSeq)) + this.projectionStore(s.sessionId).applyCached(block.values) } } else { this.listState = 'error' @@ -604,12 +602,11 @@ export class SessionManager { handleSessionAdded(summary: SessionSummary): void { this.mergeSummary(summary) this.sessions.get(summary.sessionId)?.handleBlank(summary.blank) + // A list-surface block is cached like the list response itself: the + // Session's own baseline replaces it once the client opens the Session. const projections = summary.projections if (projections !== undefined) { - const store = this.projectionStore(summary.sessionId) - for (const [key, value] of Object.entries(projections.values)) { - store.apply(key, value, sessionSeqCursor(projections.asOfSeq)) - } + this.projectionStore(summary.sessionId).applyCached(projections.values) } } diff --git a/packages/api/session-controller/src/client/sessions/projection-store.ts b/packages/api/session-controller/src/client/sessions/projection-store.ts index a53e4e73ee..24b3b22602 100644 --- a/packages/api/session-controller/src/client/sessions/projection-store.ts +++ b/packages/api/session-controller/src/client/sessions/projection-store.ts @@ -52,11 +52,18 @@ export interface ProjectionsBaseline { values: Readonly> } -/** One key's row: the latest finished value and the seq it is consistent with. */ -interface Row { - value: unknown - seq: SessionSeqCursor -} +/** + * One key's row. A `sequenced` row was computed by the Host for this Session + * within the current connection generation and carries the seq it is + * consistent with; seqs of sequenced rows are comparable with each other. A + * `cached` row was viewed from the persisted projection checkpoint by a + * header-only listing without a Session: its origin cannot vouch that its + * stored seq is comparable with the log the client later opens, so it carries + * none and yields to every sequenced write. + */ +type Row = + | { kind: 'cached'; value: unknown } + | { kind: 'sequenced'; value: unknown; seq: SessionSeqCursor } /** Per-key notification channel: the bare face plus its batching notifier. */ interface Channel { @@ -66,10 +73,14 @@ interface Channel { /** * One session's projection values. Framework semantics, uniform across every - * key: a baseline seeds rows at its cut, a push frame updates one row, and in - * both paths a lower-or-equal seq within the Host generation loses. A replayed - * frame cannot regress a value; a stale baseline cannot overwrite a newer - * frame. A key the store has never seen reads `undefined` (capability absent). Faces are identity-stable + * key. Sequenced writes (a baseline seeds rows at its cut, a push frame + * updates one row) compare seqs among themselves: a lower-or-equal seq within + * the Host generation loses, so a replayed frame cannot regress a value and a + * stale baseline cannot overwrite a newer frame. Cached writes (the session + * list's zero-I/O block) only fill keys no sequenced row holds, and a baseline + * discards every cached row before it seeds, regardless of seq: the connected + * Session is the truth and a cached value never outranks it. A key the store + * has never seen reads `undefined` (capability absent). Faces are identity-stable * per key (create-on-demand, cached) so the React side binds each exactly * once; the store-level channel (`subscribeAny`) serves coarse consumers (the * manager's list projection reads the `title` key). @@ -133,27 +144,51 @@ export class ProjectionValueStore { */ apply(key: string, value: unknown, seq: SessionSeqCursor): void { const row = this.rows.get(key) - if (row !== undefined && seq <= row.seq) return // higher seq wins; replays and stale frames drop - this.rows.set(key, { value, seq }) + // higher seq wins among sequenced rows; replays and stale frames drop. A + // cached row has no comparable seq and always yields. + if (row?.kind === 'sequenced' && seq <= row.seq) return + this.rows.set(key, { kind: 'sequenced', value, seq }) this.changed(key) } /** - * Seed from a history tail page's projections block: every carried key - * lands under the same seq rule as frames; a key the block omits is - * capability-absent as of the cut — its row clears unless a newer frame - * already superseded the cut (a stale baseline can neither overwrite nor - * clear newer values). + * Fill keys from the session list's zero-I/O cached block. A cached value + * lands only where no sequenced row exists: a connected Session has already + * answered for such a key, and the list's view of the persisted checkpoint + * cannot be newer than it. + * @param values - whole values by key viewed from the persisted checkpoint. + */ + applyCached(values: Readonly>): void { + for (const key of Object.keys(values)) { + if (this.rows.get(key)?.kind === 'sequenced') continue + this.rows.set(key, { kind: 'cached', value: values[key] }) + this.changed(key) + } + } + + /** + * Seed from a history tail page's projections block. Every cached row is + * discarded first, regardless of seq: the block comes from the connected + * Session, and a value viewed from the persisted checkpoint never outranks + * it. Then every carried key lands under the same seq rule as frames, and a + * key the block omits is capability-absent as of the cut — its row clears + * unless a newer frame already superseded the cut (a stale baseline can + * neither overwrite nor clear newer sequenced values). * @param baseline - the response's projections block. */ seed(baseline: ProjectionsBaseline): void { + for (const [key, row] of this.rows) { + if (row.kind !== 'cached') continue + this.rows.delete(key) + this.changed(key) + } // 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 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 + if (row.kind === 'sequenced' && row.seq > baseline.asOfSeq) continue this.rows.delete(key) this.changed(key) } diff --git a/packages/api/session-controller/tests/manager.client.spec.ts b/packages/api/session-controller/tests/manager.client.spec.ts index 7ee5be5397..60e96c30b9 100644 --- a/packages/api/session-controller/tests/manager.client.spec.ts +++ b/packages/api/session-controller/tests/manager.client.spec.ts @@ -203,16 +203,19 @@ describe('list lifecycle', () => { } }) - it('projects cold Session additions and an empty-cut control baseline before history exists', async ({ mock, remote }) => { + it('projects a cold Session addition as cached until an empty-cut control baseline replaces it', async ({ mock, remote }) => { const manager = makeManager(mock, remote) try { manager.handleSessionAdded({ ...summary(S1), projections: { asOfSeq: -1, values: { title: 'before history' } }, }) + expect(manager.getListSnapshot().items[0]?.title).toBe('before history') + // The control baseline is the connected Session's own value: it replaces + // the list-surface title even at the same cursor. manager.handleControlFrame({ type: 'baseline', value: { jobs: { [S1]: [] }, projections: { [S1]: { asOfSeq: -1, values: { title: 'cold baseline' } } } }, }) - expect(manager.getListSnapshot().items[0]?.title).toBe('before history') + expect(manager.getListSnapshot().items[0]?.title).toBe('cold baseline') expect(manager.getListSnapshot().jobsBySession).toEqual({}) } finally { await manager.dispose() @@ -317,23 +320,23 @@ describe('list lifecycle', () => { expect(manager.getListSnapshot().items.find(item => item.sessionId === S1)?.title).toBeUndefined() }) - it('seeds cold titles from the list rows\' projections block under higher-seq-wins', async ({ mock, remote }) => { + it('fills cold titles from the list rows\' projections block as cached values', async ({ mock, remote }) => { const manager = makeManager(mock, remote) - // A push frame landed before the list (S2's title is newer than the block's cut). + // A push frame landed before the list: S2's title is a sequenced value. manager.handleControlFrame({ type: 'projection', sessionId: S2, key: 'title', value: 'Pushed', seq: 9, }) remote.session.list.mockResolvedValue(ok({ items: [ { ...summary(S1), projections: { asOfSeq: 4, values: { title: 'Cold cached' } } }, - { ...summary(S2, { updatedAt: 200 }), projections: { asOfSeq: 5, values: { title: 'List stale' } } }, + { ...summary(S2, { updatedAt: 200 }), projections: { asOfSeq: 50, values: { title: 'List stale' } } }, ] as never[], })) await manager.refreshList() 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') - // The stale list block (seq 5) cannot overwrite the newer push frame (seq 9). + // A list block never displaces a sequenced value, whatever watermark it claims. expect(items.find(item => item.sessionId === S2)?.title).toBe('Pushed') }) diff --git a/packages/api/session-controller/tests/projection-store.client.spec.ts b/packages/api/session-controller/tests/projection-store.client.spec.ts index 2ab2bbaec6..c8af28588e 100644 --- a/packages/api/session-controller/tests/projection-store.client.spec.ts +++ b/packages/api/session-controller/tests/projection-store.client.spec.ts @@ -1,11 +1,12 @@ /** * Projection value store (push model; session-projection subsystem page: - * docs/subsystems/session-projection.md): the single - * higher-seq-wins rule on both paths (a stale baseline cannot overwrite a - * newer push frame; a replayed frame cannot regress), capability absence as - * undefined, generation invalidation, and the Session/manager wiring (tail-page - * seeding, control-stream projection routing pre- and post-instantiation, the - * list rows' title projection). + * docs/subsystems/session-projection.md): the higher-seq-wins rule among + * sequenced writes (a stale baseline cannot overwrite a newer push frame; a + * replayed frame cannot regress), cached list values yielding to every + * sequenced write regardless of seq, capability absence as undefined, + * generation invalidation, and the Session/manager wiring (tail-page seeding, + * control-stream projection routing pre- and post-instantiation, the list + * rows' title projection). */ import { describe, expect } from 'vitest' import type { SessionId } from '@deepseek-ai/dsh-api-remotes/client' @@ -75,6 +76,48 @@ describe('Session projection value semantics', () => { expect(store.get('test/marks')).toBeUndefined() }) + it('cached values fill empty keys only and never displace a sequenced row', () => { + const store = new ProjectionValueStore() + store.applyCached({ 'test/marks': { marks: ['cached'] }, title: 'Cached title' }) + expect(store.values()).toEqual({ 'test/marks': { marks: ['cached'] }, title: 'Cached title' }) + // A later cached view replaces an earlier one: neither carries a seq. + store.applyCached({ title: 'Cached again' }) + expect(store.get('title')).toBe('Cached again') + // Once a sequenced row exists, cached values for that key are ignored. + store.apply('title', 'Pushed', SessionSeq(0)) + store.applyCached({ title: 'Cached late', 'test/marks': { marks: ['cached late'] } }) + expect(store.get('title')).toBe('Pushed') + expect(store.get('test/marks')).toEqual({ marks: ['cached late'] }) + }) + + it('every sequenced write outranks a cached row regardless of seq', () => { + const store = new ProjectionValueStore() + store.applyCached({ 'test/marks': { marks: ['cached'] }, title: 'Cached title', schedule: [] }) + // A frame at the lowest cursor still replaces the cached value. + store.apply('title', 'Frame at -1', -1) + expect(store.get('title')).toBe('Frame at -1') + // A baseline discards every cached row first: the carried key lands at its + // cut, the omitted keys clear even though a cached row has no seq to compare. + store.seed({ asOfSeq: SessionSeq(2), values: { 'test/marks': { marks: ['baseline-2'] } } }) + expect(store.values()).toEqual({ 'test/marks': { marks: ['baseline-2'] } }) + // The same baseline rule keeps protecting newer sequenced rows. + store.apply('title', 'Frame at 9', SessionSeq(9)) + store.applyCached({ title: 'Cached late' }) + store.seed({ asOfSeq: SessionSeq(5), values: { 'test/marks': { marks: ['baseline-5'] } } }) + expect(store.values()).toEqual({ 'test/marks': { marks: ['baseline-5'] }, title: 'Frame at 9' }) + }) + + it('notifies faces for cached fills and for their discard by a baseline', async () => { + const store = new ProjectionValueStore() + const observed: unknown[] = [] + store.faceOf('title').subscribe(() => { observed.push(store.get('title')) }) + store.applyCached({ title: 'Cached title' }) + await Promise.resolve() + store.seed({ asOfSeq: SessionSeq(3), values: {} }) + await Promise.resolve() + expect(observed).toEqual(['Cached title', undefined]) + }) + it('clears all generation watermarks without replacing subscribed faces', async () => { const store = new ProjectionValueStore() const face = store.faceOf('test/marks') @@ -206,6 +249,38 @@ describe('manager frame routing', () => { expect(manager.getListSnapshot().items[0]?.title).toBe('Projected title') }) + it('caches a list block: a Session baseline at any cut replaces it and a list refresh cannot restore it', async ({ mock, remote }) => { + const manager = makeManager(mock, remote) + remote.session.list.mockResolvedValue(ok({ + items: [{ agentAvailable: false, + sessionId: sid('s1'), updatedAt: 1, running: false, blank: false, + // A stale record whose watermark outruns the connected Session's cut. + projections: { asOfSeq: 40, values: { title: 'Cached title', 'test/marks': { marks: ['cached'] } } }, + }], + })) + await manager.refreshList() + expect(manager.getListSnapshot().items[0]?.title).toBe('Cached title') + + // The connected Session answers at a lower cut: it still wins outright, + // and the key it omits clears rather than surviving on its stale seq. + manager.handleControlFrame({ + type: 'baseline', + value: { + jobs: {}, + projections: { [sid('s1')]: { asOfSeq: 2, values: { title: 'Connected title' } } }, + }, + }) + await Promise.resolve() + expect(manager.getListSnapshot().items[0]?.title).toBe('Connected title') + expect(manager.getListSnapshot().items[0]?.projectionValues).toEqual({ title: 'Connected title' }) + + // A later list refresh serving the stale block again cannot displace the + // sequenced value, whatever watermark the block claims. + await manager.refreshList() + expect(manager.getListSnapshot().items[0]?.title).toBe('Connected title') + expect(manager.get(sid('s1')).projections.get('test/marks')).toEqual({ marks: ['cached'] }) + }) + it('projects every retained value into list rows with stable snapshot identity', async ({ mock, remote }) => { const manager = makeManager(mock, remote) remote.session.list.mockResolvedValue(ok({