From fbebe1757ae12b5543ec235f2ede24663efe6fa5 Mon Sep 17 00:00:00 2001 From: imccyu <276526105+imccyu@users.noreply.github.com> Date: Mon, 27 Jul 2026 15:37:17 +0800 Subject: [PATCH] =?UTF-8?q?feat(gui):=20client=20projection=20cells=20?= =?UTF-8?q?=E2=80=94=20session=20dispatch=20seam,=20one-watermark=20fold,?= =?UTF-8?q?=20service=20roster?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Object layer of the session-projection RFC client base: ProjectionCellSpec/ ProjectionCell/ProjectionCellSet with the single seq-watermark rule (live and window-replace events share one filter; baseline reset re-seeds value+watermark unless a newer commit applied; absent key = capability absent), Session dispatch at appendLive/installWindow (projections block read structurally, TODO(gui) switch to the interface package), SessionsService.registerProjectionCell roster (live scopes now + future scopes at mint; disposer sweeps every session), and the provideInfo projections face (key-addressed bare cell sources). 15 object-layer specs: watermark no-rollback, late-baseline seq rule, capability absence, schema-failure degrade, duplicate-key throw, resync e2e. --- .../src/client/sessions/projection-cell.ts | 226 +++++++++++++++++ .../runtime/src/client/sessions/service.ts | 56 +++- .../runtime/src/client/sessions/session.ts | 47 +++- .../runtime/tests/projection-cell.spec.ts | 240 ++++++++++++++++++ 4 files changed, 562 insertions(+), 7 deletions(-) create mode 100644 packages/client/runtime/src/client/sessions/projection-cell.ts create mode 100644 packages/client/runtime/tests/projection-cell.spec.ts diff --git a/packages/client/runtime/src/client/sessions/projection-cell.ts b/packages/client/runtime/src/client/sessions/projection-cell.ts new file mode 100644 index 0000000000..539015992e --- /dev/null +++ b/packages/client/runtime/src/client/sessions/projection-cell.ts @@ -0,0 +1,226 @@ +/** + * Projection cells: per-session log-derived domain state on the client + * (session-projection RFC). A domain client plugin registers one cell per + * projection key at scope materialization; the framework owns the fold + * semantics — last-wins over whole-value events, guarded by a single seq + * watermark shared by the live and window-replace paths, re-seeded by the + * tail-page baseline. Cells are bare observable sources; React binding + * (useProjection) happens in web-react. + */ +import type { SessionEvent } from '@deepseek-ai/dsh-session/types' +import type { ObservableSnapshot } from '../contract/store.ts' +import { Notifier } from './notifier.ts' + +/** + * The single projection type table, typed end to end (host provider, wire + * block, client cell, React hook). Domain packages merge their keys in. + * + * TODO(gui): switch to `import type { SessionProjectionMap } from + * '@deepseek-ai/dsh-session-projection'` (pure type-only edge) once the host + * interface package lands; this placeholder is structurally identical and + * exists only because the two bases are built in parallel. No second + * client-side "views" table — one map end to end (user ruling, RFC + * Alternatives). + */ +export interface SessionProjectionMap {} + +/** + * Minimal validating-schema face (zod-compatible: `ZodType` satisfies it + * structurally). Keeps the client runtime free of a zod dependency while the + * interface package owns the real schemas. + */ +export interface ProjectionSchemaLike { + /** + * Validate a wire payload; MUST throw on mismatch. + * @param value - raw baseline payload. + * @returns the validated value. + */ + parse(value: unknown): T +} + +/** + * One domain's client-side projection contribution: the key, the wire-boundary + * schema for the baseline payload, and the whole-value event extractor. The + * signature makes delta shapes unrepresentable — `fromEvent` returns the + * complete post-change state or "not my event". + */ +export interface ProjectionCellSpec { + key: K + /** Validates the baseline payload at the wire boundary (a failed parse degrades to capability absent). */ + schema: ProjectionSchemaLike + /** + * Extract the whole post-change value from a domain event. + * @param event - any session event (live or window-replayed). + * @returns the complete value, or undefined for "not my event". + */ + fromEvent(event: SessionEvent): SessionProjectionMap[K] | undefined +} + +/** + * The fifth framework hook seat (session-projection RFC): key-addressed + * projection reader delivered through the standard kit. `undefined` uniformly + * means capability absent — host plugin unmounted, client cell unregistered, + * or no baseline landed yet. The selector overload mirrors useSession + * (per-cell uSES binding with reference-stable whole values). + */ +export type UseProjection = { + (key: K): SessionProjectionMap[K] | undefined + ( + key: K, + selector: (value: SessionProjectionMap[K] | undefined) => S, + eq?: (a: S, b: S) => boolean, + ): S +} + +/** Tail-page projections baseline (structural wire mirror; the zod schema lands with the host-base PR). */ +export interface ProjectionsBaseline { + /** The consistent-cut seq (equals the window tail seq by construction). */ + asOfSeq: number + /** Whole current values by key; a registered key absent here means the capability is absent. */ + values: Record +} + +/** Type-erased spec view the framework machinery works with (the register seam already proved the typed contract). */ +interface ErasedCellSpec { + key: string + schema: ProjectionSchemaLike + fromEvent(event: SessionEvent): unknown +} + +/** + * One key's per-session cell. Framework semantics, implemented once for all + * cells: a `lastAppliedSeq` watermark; one application rule — `event.seq > + * watermark` and `fromEvent` hit ⇒ take the whole value, raise the watermark, + * notify (microtask-batched); live and window-replace events pass the same + * filter, so replayed old pages can never roll state back; a baseline reset + * re-seeds value and watermark unless a newer commit already applied (seq + * rule); `undefined` uniformly means capability absent. + */ +export class ProjectionCell implements ObservableSnapshot { + private value: unknown = undefined + /** Highest seq whose state this cell reflects; -1 = nothing applied (pre-baseline construction state). */ + private lastAppliedSeq = -1 + /** No rebuild callback: the value is written eagerly at the application sites; the notifier only batches. */ + private readonly notifier = new Notifier(() => {}) + + /** @param spec - erased cell spec (typed at the register seam). */ + constructor(private readonly spec: ErasedCellSpec) {} + + /** + * Offer one event (live append or window replay — same filter). + * @param event - session event in log order or replayed. + */ + offerEvent(event: SessionEvent): void { + if (event.seq <= this.lastAppliedSeq) return // replay at or below the watermark: never roll back + const hit = this.spec.fromEvent(event) + if (hit === undefined) return + this.value = hit + this.lastAppliedSeq = event.seq + this.notifier.markDirty() + } + + /** + * Re-seed from a tail-page baseline. A stale baseline (cut older than an + * already-applied commit) is dropped whole — the seq rule, uniform with the + * event filter. + * @param present - whether the block carried this cell's key. + * @param raw - the key's raw wire payload (validated here; a parse failure degrades to absent). + * @param asOfSeq - the block's consistent-cut seq. + */ + resetBaseline(present: boolean, raw: unknown, asOfSeq: number): void { + if (asOfSeq < this.lastAppliedSeq) return // a newer mux commit already applied; the baseline must not overwrite it + if (present) { + try { + this.value = this.spec.schema.parse(raw) + } catch (error) { + console.error(`[web-runtime] projection baseline for "${this.spec.key}" failed validation:`, error) + this.value = undefined + } + } else { + this.value = undefined // key absent from the block: capability absent + } + this.lastAppliedSeq = asOfSeq + this.notifier.markDirty() + } + + /** + * uSES subscription entry (bare source; web-react binds the hook). + * @param listener - change callback. + * @returns the unsubscribe function. + */ + subscribe(listener: () => void): () => void { + return this.notifier.subscribe(listener) + } + + /** + * Current whole value; `undefined` means capability absent (no baseline + * carried the key, or none landed yet). + * @returns the value reference (frozen event/wire data — stable between applications). + */ + getSnapshot(): unknown { + return this.value + } +} + +/** + * The per-session cell set: registration (duplicate keys throw — one cell per + * key per session), the two dispatch entrances the Session forwards to, and + * the key-addressed read face useProjection resolves through. + */ +export class ProjectionCellSet { + private readonly cells = new Map() + + /** + * Register one cell (scope-materialization time; the caller wires the + * disposer into the scope fiber, the InputHub.shellFor pattern). + * @param spec - typed cell spec. + * @returns disposer removing the cell. + */ + register(spec: ProjectionCellSpec): () => void { + if (this.cells.has(spec.key)) throw new Error(`projection cell "${spec.key}" is already registered on this session`) + const cell = new ProjectionCell(spec as unknown as ErasedCellSpec) + this.cells.set(spec.key, cell) + return () => { + this.cells.delete(spec.key) + } + } + + /** + * Key-addressed bare source (the useProjection resolution face). + * @param key - projection key. + * @returns the cell, or undefined when no cell is registered (capability absent). + */ + cellOf(key: string): ProjectionCell | undefined { + return this.cells.get(key) + } + + /** + * Live-append dispatch (one event through every cell's filter). + * @param event - the appended live event. + */ + offerEvent(event: SessionEvent): void { + for (const cell of this.cells.values()) cell.offerEvent(event) + } + + /** + * Window-replace dispatch: every window event through the same filter — + * events newer than a cell's watermark apply, replayed old pages drop. + * @param events - the (re)installed window slice. + */ + offerWindow(events: readonly SessionEvent[]): void { + for (const event of events) this.offerEvent(event) + } + + /** + * Baseline re-seed from a tail-page response's projections block. Called + * only when the response carries the block (RFC: reset rides the block; a + * blockless response — registry-less deployment — leaves cells on the + * one-rule event path, and every un-baselined key reads absent by default). + * @param baseline - the response's projections block. + */ + resetBaseline(baseline: ProjectionsBaseline): void { + for (const [key, cell] of this.cells) { + cell.resetBaseline(Object.hasOwn(baseline.values, key), baseline.values[key], baseline.asOfSeq) + } + } +} diff --git a/packages/client/runtime/src/client/sessions/service.ts b/packages/client/runtime/src/client/sessions/service.ts index ec8ddc2354..0b765040f5 100644 --- a/packages/client/runtime/src/client/sessions/service.ts +++ b/packages/client/runtime/src/client/sessions/service.ts @@ -26,6 +26,7 @@ import { createScope, scopeOf as scopeTagOf } from '../agents/scope.ts' import { SessionManager } from './manager.ts' import type { SessionListPhase } from './manager.ts' import type { Session } from './session.ts' +import type { ProjectionCellSpec, SessionProjectionMap } from './projection-cell.ts' /** Session list row projected from the host list RPC plus live stream increments. */ export interface SessionSummary { @@ -165,6 +166,15 @@ export class SessionsService { private readonly scopes = new Map() /** Registered per-session standard-props providers, in registration order. */ private readonly providers: SessionProvideDescriptor[] = [] + /** + * Projection-cell roster (session-projection RFC): each registered spec is + * applied to every live scope's session and to every future scope at mint. + * The per-spec map tracks live-session disposers so a provider unload (HMR) + * removes its cell from every session; scope drop just forgets the row (the + * Session instance dies with the scope). + */ + private readonly projectionCells = + new Map, Map void>>() /** Static no-session projection, rebuilt only when the provider roster changes. */ private maybeInfo: SessionMaybeProvideInfo /** @@ -232,6 +242,29 @@ export class SessionsService { } } + /** + * Register a projection cell spec (session-projection RFC): the framework + * materializes one cell per session — on every already-live scope now, and + * on every future scope at mint (the binding-fed shellFor timing) — and the + * cell set dies with the scope. One registration per domain; duplicate keys + * fail loud at materialization. + * @param spec - typed cell spec (key + wire schema + whole-value extractor). + * @returns disposer removing the spec from the roster and its cell from every live session. + */ + registerProjectionCell(spec: ProjectionCellSpec): () => void { + const erased = spec as ProjectionCellSpec + const disposers = new Map void>() + this.projectionCells.set(erased, disposers) + for (const record of this.scopes.values()) { + disposers.set(record.binding.sessionId, record.binding.session.projections.register(erased)) + } + return () => { + this.projectionCells.delete(erased) + for (const dispose of disposers.values()) dispose() + disposers.clear() + } + } + /** Rebuild every live scope's standard-props bundle after a provider roster change. */ private rematerializeProvideBundles(): void { this.maybeInfo = this.materializeMaybeProvideInfo() @@ -254,7 +287,7 @@ export class SessionsService { props[name] = undefined } } - return { sessionId: undefined, hooks, props } + return { sessionId: undefined, hooks, props } // no projections face: every key reads absent without a session } /** Materialize the standard-props bundle for one session (fails loud on duplicate member names). */ @@ -287,7 +320,14 @@ export class SessionsService { props[name] = contributedProps[name] } } - return { sessionId: binding.sessionId, hooks, props } + return { + sessionId: binding.sessionId, + hooks, + props, + // The useProjection seat: key-addressed bare cell sources off the + // session's cell set (open key space — never a static roster member). + projections: { cellOf: key => binding.session.projections.cellOf(key) }, + } } /** @@ -485,6 +525,12 @@ export class SessionsService { // The Session owns its scoped dispatch point (host Agent.loopCtx mirror); // mint and bind are one step so a live scope record implies a bound actx. session.bindScope(ctx) + // Materialize the projection-cell roster on the freshly scoped session + // (dropScope swept the previous scope's rows, so a re-mint registers on + // whatever instance the manager now holds — fresh or resident). + for (const [spec, disposers] of this.projectionCells) { + disposers.set(id, session.projections.register(spec)) + } const binding: SessionBinding = { sessionId: id, session, ctx } const record: ScopeRecord = { fiber, @@ -559,6 +605,12 @@ export class SessionsService { // Release the Session's dispatch point with the scope it belongs to (a // surviving instance — the live Intent — rebinds when resolve re-mints). record.binding.session.unbindScope() + // Sweep the projection-cell rows with the scope (instance and scope share + // one lifecycle; a re-mint re-registers the roster on the new instance). + for (const disposers of this.projectionCells.values()) { + disposers.get(id)?.() + disposers.delete(id) + } // Optional lookup: slots and sessions are sibling services with no // declared dependency; a slots-less boot (object-layer tests) skips. this.rootCtx.get('slots')?.pruneStoreScope(id) diff --git a/packages/client/runtime/src/client/sessions/session.ts b/packages/client/runtime/src/client/sessions/session.ts index 1dd0283429..a77197e7e3 100644 --- a/packages/client/runtime/src/client/sessions/session.ts +++ b/packages/client/runtime/src/client/sessions/session.ts @@ -20,6 +20,8 @@ import { PendingWait } from './pending.ts' import { FoldAdapter } from './fold-adapter.ts' import { Notifier } from './notifier.ts' import { PartialAccumulator } from './partial.ts' +import { ProjectionCellSet } from './projection-cell.ts' +import type { ProjectionsBaseline } from './projection-cell.ts' /** Messages requested per history page. */ export const PAGE_MESSAGES = 50 @@ -126,6 +128,17 @@ export class Session implements ObservableSnapshot { /** subscribed.lastSeq baseline (gap detection; null when no subscribed frame arrived — degrade to the liveBuffer dedup path). */ private subscribedLastSeq: number | null = null + /** + * Per-session projection cells (session-projection RFC): domain client + * plugins register cells at scope materialization (disposer rides the scope + * fiber, the InputHub.shellFor pattern); the Session dispatches its two + * event entrances — appendLive (live signal) and installWindow (window + * replace + baseline reset) — into the set. Cells are read via + * `projections.cellOf(key)` (the useProjection resolution face); the + * conversation snapshot never carries projection values. + */ + readonly projections = new ProjectionCellSet() + private snapshotCache: ConversationSnapshot private readonly notifier = new Notifier(() => { this.snapshotCache = this.buildSnapshot() @@ -482,13 +495,13 @@ export class Session implements ObservableSnapshot { this.openError = result.error return } - this.installWindow(result.value.events, result.value.hasMore, result.value.todos) + this.installWindow(result.value.events, result.value.hasMore, result.value.todos, projectionsOf(result.value)) // Gap detection (§D.3-4): baseline past the window tail and liveBuffer did not cover it -> pull the tail page once more. const tailSeq = this.windowTailSeq() if (this.subscribedLastSeq !== null && tailSeq !== null && this.subscribedLastSeq > tailSeq) { result = (await this.api.sessions.history({ sessionId: this.sessionId, maxMessages: PAGE_MESSAGES })).result if (generation !== this.openGeneration) return - if (result.ok) this.installWindow(result.value.events, result.value.hasMore, result.value.todos) + if (result.ok) this.installWindow(result.value.events, result.value.hasMore, result.value.todos, projectionsOf(result.value)) } this.openState = 'open' } catch (error) { @@ -505,8 +518,12 @@ export class Session implements ObservableSnapshot { /** Install the history window + stitch the liveBuffer (seq is the sole dedup key). * Stitching MUST NOT route through acceptLiveEvent: openState is still 'loading' here * (doOpen flips it after install), so recursing would push every buffered event straight - * back into liveBuffer where nothing ever drains it — a silent drop loop (audit S1). */ - private installWindow(entries: HistoryEntry[], hasMore: boolean, todos: readonly TodoItem[] | undefined): void { + * back into liveBuffer where nothing ever drains it — a silent drop loop (audit S1). + * Projection dispatch (window-replace signal): a carried projections block re-seeds every + * cell first (value + watermark, seq-rule guarded), then the window events pass the same + * per-cell filter as live appends — a blockless response leaves cells folding from events + * alone, and replayed pages can never roll a cell back. */ + private installWindow(entries: HistoryEntry[], hasMore: boolean, todos: readonly TodoItem[] | undefined, projections?: ProjectionsBaseline): void { this.events = entries.map(e => e.event) this.views = entries.map(e => e.view) this.baseSeq = this.events[0]?.seq ?? 0 @@ -521,6 +538,8 @@ export class Session implements ObservableSnapshot { this.todos = todos ?? [] this.foldAdapter.reset(this.events, this.baseSeq, this.views) this.rebuildDerivedFromWindow() + if (projections !== undefined) this.projections.resetBaseline(projections) + this.projections.offerWindow(this.events) const buffered = this.liveBuffer this.liveBuffer = [] for (const item of buffered) this.appendLive(item.event, item.view) @@ -535,6 +554,8 @@ export class Session implements ObservableSnapshot { this.views.push(view) this.foldAdapter.append(event, view) this.applyEventSideEffects(event, view) + // Projection dispatch (live signal): same filter as the window path. + this.projections.offerEvent(event) } /** Land a live session/event (open/repair in flight -> buffer; overlapping seq -> drop; @@ -569,7 +590,7 @@ export class Session implements ObservableSnapshot { const { result } = await this.api.sessions.history({ sessionId: this.sessionId, maxMessages: PAGE_MESSAGES }) // Failure or superseded by a full resync: drop — the resync path rebuilds and clears the buffer itself. if (result.ok && generation === this.openGeneration && this.openState === 'open') { - this.installWindow(result.value.events, result.value.hasMore, result.value.todos) + this.installWindow(result.value.events, result.value.hasMore, result.value.todos, projectionsOf(result.value)) } } catch (error) { console.error('[web-runtime] gap repair failed:', error) @@ -829,3 +850,19 @@ function derivePhase(hasContent: boolean, promptAttempted: boolean): ComposerPha if (hasContent) return 'active' return promptAttempted ? 'engaging' : 'blank' } + +/** + * Structural read of the optional projections block on a history response. + * TODO(gui): drop this narrowing once the host-base PR (dsh-session-projection + * + apiproxy block) lands and the wire type carries `projections` — parallel + * construction posture, same as the code-dispatch event narrowing above. + * @param value - the history response value. + * @returns the block, or undefined (loadOlder pages and blockless deployments). + */ +function projectionsOf(value: object): ProjectionsBaseline | undefined { + const block = (value as { projections?: ProjectionsBaseline }).projections + if (block === undefined) return undefined + return typeof block.asOfSeq === 'number' && typeof block.values === 'object' && block.values !== null + ? block + : undefined +} diff --git a/packages/client/runtime/tests/projection-cell.spec.ts b/packages/client/runtime/tests/projection-cell.spec.ts new file mode 100644 index 0000000000..78a661222b --- /dev/null +++ b/packages/client/runtime/tests/projection-cell.spec.ts @@ -0,0 +1,240 @@ +/** + * Projection cells (session-projection RFC): the one watermark rule shared by + * live and window paths (replayed pages never roll back), baseline reset + * semantics (late baseline never overwrites a newer commit), capability + * absence as undefined, and the Session/SessionsService dispatch wiring. + */ +import { Context } from 'cordis' +import { describe, expect, it } from 'vitest' +import type { SessionEvent } from '@deepseek-ai/dsh-session/types' +import type { SessionId } from '@deepseek-ai/dsh-client-connection/client' +import { ProjectionCellSet } from '../src/client/sessions/projection-cell.ts' +import type { ProjectionCellSpec } from '../src/client/sessions/projection-cell.ts' +import { Session } from '../src/client/sessions/session.ts' +import { SessionsService } from '../src/client/sessions/service.ts' +import { FakeApiClient, ok } from './fake-api.ts' +import { entries, plainTurn } from './event-script.ts' + +// Test-domain key merged into the (placeholder) projection map: a whole-value +// marker list, the smallest last-wins shape. +declare module '../src/client/sessions/projection-cell.ts' { + interface SessionProjectionMap { + 'test/marks': { marks: string[] } + } +} + +const SID = 'fk-s1' as SessionId + +/** Whole-value domain event carrying the complete post-change state. */ +const markEvent = (seq: number, marks: string[]): SessionEvent => + ({ seq, time: 1_700_000_000_000 + seq, type: 'test/mark', data: { marks } }) as unknown as SessionEvent + +/** Loose schema: passes objects with a marks array through, throws otherwise. */ +const marksSpec = (): ProjectionCellSpec<'test/marks'> => ({ + key: 'test/marks', + schema: { + parse: (value) => { + if (typeof value === 'object' && value !== null && Array.isArray((value as { marks?: unknown }).marks)) { + return value as { marks: string[] } + } + throw new Error('not a marks payload') + }, + }, + fromEvent: (event) => ((event.type as string) === 'test/mark' + ? (event as unknown as { data: { marks: string[] } }).data + : undefined), +}) + +describe('ProjectionCellSet semantics', () => { + function bench() { + const set = new ProjectionCellSet() + const dispose = set.register(marksSpec()) + const cell = set.cellOf('test/marks') + if (cell === undefined) throw new Error('cell missing after register') + return { set, cell, dispose } + } + + it('starts absent (undefined) until any signal lands', () => { + const { cell } = bench() + expect(cell.getSnapshot()).toBeUndefined() + }) + + it('applies whole values last-wins by seq and never rolls back on replayed old events', () => { + const { set, cell } = bench() + set.offerEvent(markEvent(5, ['a'])) + set.offerEvent(markEvent(9, ['a', 'b'])) + expect(cell.getSnapshot()).toEqual({ marks: ['a', 'b'] }) + // A replayed old page (window path) passes the same filter and drops. + set.offerWindow([markEvent(3, ['stale']), markEvent(9, ['a', 'b'])]) + expect(cell.getSnapshot()).toEqual({ marks: ['a', 'b'] }) + }) + + it('re-seeds value and watermark from a baseline, and events at or below asOfSeq drop after it', () => { + const { set, cell } = bench() + set.resetBaseline({ asOfSeq: 20, values: { 'test/marks': { marks: ['x'] } } }) + expect(cell.getSnapshot()).toEqual({ marks: ['x'] }) + set.offerEvent(markEvent(18, ['older-than-cut'])) + expect(cell.getSnapshot()).toEqual({ marks: ['x'] }) + set.offerEvent(markEvent(21, ['newer'])) + expect(cell.getSnapshot()).toEqual({ marks: ['newer'] }) + }) + + it('drops a late baseline whose cut predates an already-applied commit (seq rule)', () => { + const { set, cell } = bench() + set.offerEvent(markEvent(30, ['live-commit'])) + set.resetBaseline({ asOfSeq: 25, values: { 'test/marks': { marks: ['stale-baseline'] } } }) + expect(cell.getSnapshot()).toEqual({ marks: ['live-commit'] }) + }) + + it('marks a key absent when the block omits it — capability absence is undefined', () => { + const { set, cell } = bench() + set.offerEvent(markEvent(5, ['a'])) + set.resetBaseline({ asOfSeq: 10, values: {} }) + expect(cell.getSnapshot()).toBeUndefined() + }) + + it('degrades a baseline payload failing schema validation to absent instead of poisoning the cell', () => { + const { set, cell } = bench() + set.resetBaseline({ asOfSeq: 10, values: { 'test/marks': 'not-an-object' } }) + expect(cell.getSnapshot()).toBeUndefined() + // The watermark still advanced to the cut: pre-cut events stay dropped. + set.offerEvent(markEvent(8, ['pre-cut'])) + expect(cell.getSnapshot()).toBeUndefined() + }) + + it('throws on duplicate key registration and frees the key through the disposer', () => { + const { set, dispose } = bench() + expect(() => set.register(marksSpec())).toThrow(/already registered/) + dispose() + expect(set.cellOf('test/marks')).toBeUndefined() + expect(() => set.register(marksSpec())).not.toThrow() + }) + + it('notifies subscribers on application (microtask-batched) and not on filtered events', async () => { + const { set, cell } = bench() + let ticks = 0 + cell.subscribe(() => { ticks += 1 }) + set.offerEvent(markEvent(5, ['a'])) + await Promise.resolve() + expect(ticks).toBe(1) + set.offerEvent(markEvent(3, ['replay'])) + set.offerEvent({ seq: 6, time: 6, type: 'unrelated/event', data: {} } as unknown as SessionEvent) + await Promise.resolve() + expect(ticks).toBe(1) + }) +}) + +describe('Session dispatch wiring', () => { + function makeSession() { + const api = new FakeApiClient() + const session = new Session(SID, api) + const dispose = session.projections.register(marksSpec()) + const cell = session.projections.cellOf('test/marks') + if (cell === undefined) throw new Error('cell missing after register') + return { api, session, cell, dispose } + } + + it('feeds live appends through the cell filter', async () => { + const { api, session, cell } = makeSession() + api.onHistory = () => Promise.resolve(ok({ events: entries(plainTurn(0, 0, '问', '答')) as never[], hasMore: false })) + await session.open() + session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event: markEvent(6, ['live']) }) + expect(cell.getSnapshot()).toEqual({ marks: ['live'] }) + }) + + it('re-seeds from a history response carrying a projections block, then folds newer window events', async () => { + const { api, session, cell } = makeSession() + const window = [...plainTurn(0, 0, '问', '答'), markEvent(6, ['from-window'])] + api.onHistory = () => Promise.resolve(ok({ + events: entries(window) as never[], hasMore: false, + projections: { asOfSeq: 4, values: { 'test/marks': { marks: ['from-baseline'] } } }, + } as never)) + await session.open() + // Baseline cut at 4; the window's seq-6 domain event is newer and wins. + expect(cell.getSnapshot()).toEqual({ marks: ['from-window'] }) + }) + + it('treats a blockless response as event-only folding (no reset), and a resync repull cannot roll back', async () => { + const { api, session, cell } = makeSession() + api.onHistory = () => Promise.resolve(ok({ events: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false })) + await session.open() + expect(cell.getSnapshot()).toBeUndefined() + session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event: markEvent(6, ['live']) }) + expect(cell.getSnapshot()).toEqual({ marks: ['live'] }) + // Reconnect resync repulls the same window (no block, no domain events): state holds. + await session.resync() + expect(cell.getSnapshot()).toEqual({ marks: ['live'] }) + }) + + it('applies the stale-baseline guard end to end: a resync whose block predates a live commit keeps the commit', async () => { + const { api, session, cell } = makeSession() + api.onHistory = () => Promise.resolve(ok({ + events: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false, + projections: { asOfSeq: 5, values: { 'test/marks': { marks: ['baseline'] } } }, + } as never)) + await session.open() + expect(cell.getSnapshot()).toEqual({ marks: ['baseline'] }) + // Contiguous live commit applies immediately (seq 6 = tail 5 + 1)… + session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event: markEvent(6, ['commit-6']) }) + expect(cell.getSnapshot()).toEqual({ marks: ['commit-6'] }) + // …then a resync repull serves the same stale block (cut 5 < applied 6): + // the baseline reset must not overwrite the newer commit (seq rule). + await session.resync() + expect(cell.getSnapshot()).toEqual({ marks: ['commit-6'] }) + }) +}) + +describe('SessionsService roster', () => { + const sid = (s: string): SessionId => s as SessionId + + async function bench() { + const ctx = new Context() + const api = new FakeApiClient() + const svc = new SessionsService(ctx, api) + api.onList = () => Promise.resolve(ok({ + items: [{ sessionId: sid('s1'), updatedAt: 1, running: false, blank: false }], + }) as never) + await svc.refresh() + await Promise.resolve() + return { ctx, api, svc } + } + + it('materializes registered specs on already-live scopes and future scopes alike', async () => { + const b = await bench() + const binding1 = b.svc.binding(sid('s1')) + if (binding1 === undefined) throw new Error('no binding for s1') + b.svc.registerProjectionCell(marksSpec()) + expect(binding1.session.projections.cellOf('test/marks')).toBeDefined() + // A session arriving later gets the roster at scope mint. + b.api.onList = () => Promise.resolve(ok({ + items: [ + { sessionId: sid('s1'), updatedAt: 1, running: false, blank: false }, + { sessionId: sid('s2'), updatedAt: 2, running: false, blank: false }, + ], + }) as never) + await b.svc.refresh() + await Promise.resolve() + const binding2 = b.svc.binding(sid('s2')) + expect(binding2?.session.projections.cellOf('test/marks')).toBeDefined() + }) + + it('exposes the key-addressed cell face on provideInfo (the useProjection resolution path)', async () => { + const b = await bench() + b.svc.registerProjectionCell(marksSpec()) + const info = b.svc.provideInfo('s1') + if (info === undefined) throw new Error('no provide info for s1') + expect(info.projections?.cellOf('test/marks')).toBeDefined() + expect(info.projections?.cellOf('test/ghost')).toBeUndefined() + // The no-session projection carries no face: every key reads absent. + expect(b.svc.maybeProvideInfo(undefined).projections).toBeUndefined() + }) + + it('removes the cell from every live session through the disposer (HMR semantics)', async () => { + const b = await bench() + const dispose = b.svc.registerProjectionCell(marksSpec()) + const binding = b.svc.binding(sid('s1')) + expect(binding?.session.projections.cellOf('test/marks')).toBeDefined() + dispose() + expect(binding?.session.projections.cellOf('test/marks')).toBeUndefined() + }) +})