From 913125294b734d9e1d88c0778ac67ec3de4c3257 Mon Sep 17 00:00:00 2001 From: imccyu <276526105+imccyu@users.noreply.github.com> Date: Mon, 27 Jul 2026 22:02:10 +0800 Subject: [PATCH] refactor(gui): retire the client-side projection cell machinery (zero shim) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Client-side domain folding is gone (RFC final: the host is the only computation site): ProjectionCellSpec/fromEvent, ProjectionCellSet, the SessionsService cell roster, and the Session event-dispatch projection hooks all delete; Session.projections becomes the generic value store (manager-owned via SessionOptions so frames landing before instantiation and the history baseline converge on one row set), and installWindow only seeds the store from a carried block. The cell specs retire with the machinery — the value store's own spec owns the seq semantics now. --- packages/client/runtime/src/client/index.ts | 11 +- .../src/client/sessions/projection-cell.ts | 230 ----------------- .../runtime/src/client/sessions/service.ts | 51 +--- .../runtime/src/client/sessions/session.ts | 72 +++--- .../runtime/tests/projection-cell.spec.ts | 242 ------------------ .../runtime/tests/projection-todo.spec.ts | 92 ------- 6 files changed, 37 insertions(+), 661 deletions(-) delete mode 100644 packages/client/runtime/src/client/sessions/projection-cell.ts delete mode 100644 packages/client/runtime/tests/projection-cell.spec.ts delete mode 100644 packages/client/runtime/tests/projection-todo.spec.ts diff --git a/packages/client/runtime/src/client/index.ts b/packages/client/runtime/src/client/index.ts index 3b1d11140a..a7e9104b6c 100644 --- a/packages/client/runtime/src/client/index.ts +++ b/packages/client/runtime/src/client/index.ts @@ -7,7 +7,7 @@ import { SessionsService } from './sessions/service.ts' import type { SessionListState } from './sessions/service.ts' import { WorkspacesService } from './workspaces/service.ts' import type { ConversationSnapshot, RunningToolCall, ToolResultNode } from './sessions/conversation.ts' -import type { UseProjection } from './sessions/projection-cell.ts' +import type { UseProjection } from './sessions/projection-store.ts' export { SlotsService } from './slots.ts' export type { RootOwnerProps } from './slots.ts' @@ -35,12 +35,11 @@ export type { } from './sessions/conversation.ts' export { PendingWait } from './sessions/pending.ts' export type { PendingInteraction, PendingKind, PendingPayloads } from './sessions/pending.ts' -// Projection cells (session-projection RFC): domain plugins register cells at -// scope materialization via `binding.session.projections.register(spec)`. +// Projection value store (session-projection RFC, push model): host-computed +// whole values per key; domains ship projection support with zero client code. export type { - ProjectionCell, ProjectionCellSet, ProjectionCellSpec, ProjectionSchemaLike, ProjectionsBaseline, - SessionProjectionMap, UseProjection, -} from './sessions/projection-cell.ts' + ProjectionsBaseline, ProjectionValueStore, SessionProjectionMap, UseProjection, +} from './sessions/projection-store.ts' export type { SessionId } from '@deepseek-ai/dsh-client-connection/client' /** Client-side Cordis context after declaration merging. */ diff --git a/packages/client/runtime/src/client/sessions/projection-cell.ts b/packages/client/runtime/src/client/sessions/projection-cell.ts deleted file mode 100644 index b4b5204416..0000000000 --- a/packages/client/runtime/src/client/sessions/projection-cell.ts +++ /dev/null @@ -1,230 +0,0 @@ -/** - * 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 { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection/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) — the interface package's pure-type outlet -// (`/types`, zero imports), never the package root: the root's dsh-agent → -// dsh-session chain would drag the host `Context.sessions` merge into the -// client program (one program must not hold both sides). No second -// client-side "views" table (user ruling, RFC Alternatives). -export type { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection/types' - -/** - * 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 — structurally identical to the wire's - * `SessionProjectionsBlock` (apiproxy api layer), restated here so the - * React-free cell framework depends only on the type table, not the wire - * package's response vocabulary. - */ -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: Partial -} - -/** 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 { - // Erased view: the framework walks the open key space; per-key typing - // lives at the cell spec seam (schema.parse re-establishes it). - const values = baseline.values as Record - for (const [key, cell] of this.cells) { - cell.resetBaseline(Object.hasOwn(values, key), 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 0b765040f5..17c4db3a85 100644 --- a/packages/client/runtime/src/client/sessions/service.ts +++ b/packages/client/runtime/src/client/sessions/service.ts @@ -26,7 +26,6 @@ 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 { @@ -166,15 +165,6 @@ 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 /** @@ -242,29 +232,6 @@ 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() @@ -324,9 +291,9 @@ export class SessionsService { 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) }, + // The useProjection seat: key-addressed bare value faces off the + // session's projection store (open key space — never a static roster member). + projections: { faceOf: key => binding.session.projections.faceOf(key) }, } } @@ -525,12 +492,6 @@ 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, @@ -605,12 +566,6 @@ 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 ec5c9c6023..c47a0ae43c 100644 --- a/packages/client/runtime/src/client/sessions/session.ts +++ b/packages/client/runtime/src/client/sessions/session.ts @@ -2,7 +2,7 @@ import type { Context } from 'cordis' import type { ContentBlock } from '@deepseek-ai/dsh-llm/types' -import type { SessionEvent, TodoItem } from '@deepseek-ai/dsh-session/types' +import type { SessionEvent } from '@deepseek-ai/dsh-session/types' import type { HistoryEntry, IApiClient, MuxFrame, RpcError, RpcId, RpcResult, SessionId, ToolEventView, @@ -20,8 +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' +import { ProjectionValueStore } from './projection-store.ts' +import type { ProjectionsBaseline } from './projection-store.ts' /** Messages requested per history page. */ export const PAGE_MESSAGES = 50 @@ -37,6 +37,12 @@ export interface SessionOptions { * (hidden, still reusable by connectWorkspace). */ onEngaged?(session: Session): void + /** + * Manager-owned projection value store to adopt (frames route through the + * manager and values outlive instantiation); omitted, the Session owns a + * private store (bare object-layer construction). + */ + projections?: ProjectionValueStore } /** Queue-row preview cap: the dock renders one line, the full content never leaves the host mirror. */ @@ -101,9 +107,6 @@ export class Session implements ObservableSnapshot { private queueCache: { rev: number; value: QueuedMessage[] } | null = null private frozenRev = 0 private nodesCache: { folded: readonly ConversationNode[]; frozenRev: number; value: readonly ConversationNode[] } | null = null - /** Current whole-list todo/write projection: each tail history response replaces it (an omitted - * field is the authoritative empty list) and every live write overwrites it. */ - private todos: readonly TodoItem[] = [] /** `run_code` sub-dispatches by parent callId (window-derived, like openCalls). Appends * copy-on-write the per-parent array so published snapshot references never mutate. */ private codeDispatches = new Map() @@ -129,15 +132,17 @@ export class Session implements ObservableSnapshot { 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. + * Per-session projection value store (session-projection RFC, push model): + * finished whole values computed on the host, seeded by the tail page's + * projections block and updated by `session/projection` frames under the + * one higher-seq-wins rule. 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 + * the store outlives instantiation, the title-snapshot precedent); a bare + * construction gets a private store. */ - readonly projections = new ProjectionCellSet() + readonly projections: ProjectionValueStore private snapshotCache: ConversationSnapshot private readonly notifier = new Notifier(() => { @@ -162,6 +167,7 @@ export class Session implements ObservableSnapshot { private readonly api: IApiClient, private readonly options: SessionOptions = {}, ) { + this.projections = options.projections ?? new ProjectionValueStore() this.snapshotCache = this.buildSnapshot() } @@ -495,13 +501,13 @@ export class Session implements ObservableSnapshot { this.openError = result.error return } - this.installWindow(result.value.events, result.value.hasMore, result.value.todos, result.value.projections) + this.installWindow(result.value.events, result.value.hasMore, result.value.projections) // 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, result.value.projections) + if (result.ok) this.installWindow(result.value.events, result.value.hasMore, result.value.projections) } this.openState = 'open' } catch (error) { @@ -519,27 +525,17 @@ export class Session implements ObservableSnapshot { * 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). - * 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 { + * A carried projections block seeds the value store (higher seq wins, so a stale + * baseline cannot overwrite a newer push frame); the window events themselves are + * never folded — the host is the only computation site. */ + private installWindow(entries: HistoryEntry[], hasMore: boolean, projections?: ProjectionsBaseline): void { this.events = entries.map(e => e.event) this.views = entries.map(e => e.view) this.baseSeq = this.events[0]?.seq ?? 0 this.hasMore = hasMore - // Session-level projection from the tail page (full-log latest todo/write, - // independent of the window); an in-window write below re-derives the same - // value, and later live events keep overwriting it. Every caller here is a - // tail request (no beforeSeq), which the host answers with the projection - // or omits it only when the full log holds no todo/write — so an absent - // field is the authoritative empty list, not a missing carrier. Assigning - // it clears a plan the log never kept (a write lost to a host crash). - 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) + if (projections !== undefined) this.projections.seed(projections) const buffered = this.liveBuffer this.liveBuffer = [] for (const item of buffered) this.appendLive(item.event, item.view) @@ -554,8 +550,6 @@ 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; @@ -590,7 +584,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, result.value.projections) + this.installWindow(result.value.events, result.value.hasMore, result.value.projections) } } catch (error) { console.error('[web-runtime] gap repair failed:', error) @@ -710,10 +704,6 @@ export class Session implements ObservableSnapshot { if (this.openCalls.delete(String(event.data.callId))) this.callsRev++ return } - case 'todo/write': { - this.todos = event.data.todos - return - } case 'turn/end': { // Aborted turns never finalize. The accumulated partial is VALUE, not residue: freeze it // into an interrupted terminal node (pulse stops, text survives) instead of deleting it. @@ -758,10 +748,7 @@ export class Session implements ObservableSnapshot { /** Re-derive state (partial/openCalls/frozenNodes) from raw window events after a rebuild — keeps * paging/stitching consistent, and makes the live freeze and the history replay converge on the - * same interrupted nodes (chunks are logged, so the replayed sweep re-freezes identical text). - * todos is deliberately NOT reset: it is session-level (seeded by the tail page's full-log - * projection, not derivable from an arbitrary window). The window always extends to the log - * tail, so an in-window todo/write can only overwrite it with the same latest value. */ + * same interrupted nodes (chunks are logged, so the replayed sweep re-freezes identical text). */ private rebuildDerivedFromWindow(): void { this.partial = null this.openCalls.clear() @@ -831,7 +818,6 @@ export class Session implements ObservableSnapshot { promptError: this.promptError, blank: this.blankBit, lastAgentError: this.lastAgentError, - todos: this.todos, } } } diff --git a/packages/client/runtime/tests/projection-cell.spec.ts b/packages/client/runtime/tests/projection-cell.spec.ts deleted file mode 100644 index 85609d20a7..0000000000 --- a/packages/client/runtime/tests/projection-cell.spec.ts +++ /dev/null @@ -1,242 +0,0 @@ -/** - * 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 projection map (the interface package's -// pure-type outlet): a whole-value marker list, the smallest last-wins shape. -declare module '@deepseek-ai/dsh-session-projection/types' { - 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() - // Deliberately malformed wire payload: the typed block cannot express it, - // which is exactly why the boundary schema exists. - set.resetBaseline({ asOfSeq: 10, values: { 'test/marks': 'not-an-object' as never } }) - 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() - }) -}) diff --git a/packages/client/runtime/tests/projection-todo.spec.ts b/packages/client/runtime/tests/projection-todo.spec.ts deleted file mode 100644 index 8b98a82edf..0000000000 --- a/packages/client/runtime/tests/projection-todo.spec.ts +++ /dev/null @@ -1,92 +0,0 @@ -/** - * Knife-4 acceptance probe (session-projection RFC): the todo domain's client - * cell — `fromEvent: todo/write ⇒ whole list` — runs end to end on the - * UNMODIFIED cell framework: baseline seeding from a history response's - * projections block, live last-wins folding, and the seq guard, with the - * `todos` key merged test-locally the same way the domain client plugin will - * (through the interface package's pure-type outlet). Zero framework edits. - */ -import { describe, expect, it } from 'vitest' -import type { SessionEvent } from '@deepseek-ai/dsh-session/types' -import type { TodoItem } from '@deepseek-ai/dsh-session/types' -import type { SessionId } from '@deepseek-ai/dsh-client-connection/client' -import type { ProjectionCellSpec } from '../src/client/sessions/projection-cell.ts' -import { Session } from '../src/client/sessions/session.ts' -import { FakeApiClient, ok } from './fake-api.ts' -import { entries, plainTurn } from './event-script.ts' - -declare module '@deepseek-ai/dsh-session-projection/types' { - interface SessionProjectionMap { - todos: TodoItem[] | null - } -} - -const SID = 'fk-todo' as SessionId - -const todoEvent = (seq: number, todos: TodoItem[]): SessionEvent => - ({ seq, time: 1_700_000_000_000 + seq, type: 'todo/write', data: { todos } }) as unknown as SessionEvent - -/** The exact cell the todo domain client plugin will register: whole-list fromEvent, array-or-null schema. */ -const todosSpec = (): ProjectionCellSpec<'todos'> => ({ - key: 'todos', - schema: { - parse: (value) => { - if (value === null || Array.isArray(value)) return value as TodoItem[] | null - throw new Error('not a todos payload') - }, - }, - fromEvent: event => (event.type === 'todo/write' - ? (event as unknown as { data: { todos: TodoItem[] } }).data.todos - : undefined), -}) - -function makeSession() { - const api = new FakeApiClient() - const session = new Session(SID, api) - session.projections.register(todosSpec()) - const cell = session.projections.cellOf('todos') - if (cell === undefined) throw new Error('cell missing after register') - return { api, session, cell } -} - -describe('todo projection cell over the unmodified framework', () => { - it('seeds null from a pre-first-write baseline, then a live todo/write replaces it whole', async () => { - const { api, session, cell } = makeSession() - api.onHistory = () => Promise.resolve(ok({ - events: entries(plainTurn(0, 0, 'q', 'a')) as never[], hasMore: false, - projections: { asOfSeq: 5, values: { todos: null } }, - } as never)) - await session.open() - expect(cell.getSnapshot()).toBeNull() - const list: TodoItem[] = [{ content: 'ship knife 4', status: 'in_progress' }] - session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event: todoEvent(6, list) }) - expect(cell.getSnapshot()).toEqual(list) - }) - - it('seeds the whole list from the baseline and drops a replayed older snapshot (last-wins)', async () => { - const { api, session, cell } = makeSession() - const current: TodoItem[] = [ - { content: 'a', status: 'completed' }, - { content: 'b', status: 'pending' }, - ] - api.onHistory = () => Promise.resolve(ok({ - events: entries(plainTurn(0, 0, 'q', 'a')) as never[], hasMore: false, - projections: { asOfSeq: 9, values: { todos: current } }, - } as never)) - await session.open() - expect(cell.getSnapshot()).toEqual(current) - // A replayed pre-cut write (window path) must not roll the list back. - session.projections.offerWindow([todoEvent(4, [{ content: 'stale', status: 'pending' }])]) - expect(cell.getSnapshot()).toEqual(current) - }) - - it('reads capability-absent (undefined) when the block omits the todos key', async () => { - const { api, session, cell } = makeSession() - api.onHistory = () => Promise.resolve(ok({ - events: entries(plainTurn(0, 0, 'q', 'a')) as never[], hasMore: false, - projections: { asOfSeq: 5, values: {} }, - } as never)) - await session.open() - expect(cell.getSnapshot()).toBeUndefined() - }) -})