feat(gui): client projection cells — session dispatch seam, one-watermark fold, service roster
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.
This commit is contained in:
226
packages/client/runtime/src/client/sessions/projection-cell.ts
Normal file
226
packages/client/runtime/src/client/sessions/projection-cell.ts
Normal file
@@ -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<T>` satisfies it
|
||||||
|
* structurally). Keeps the client runtime free of a zod dependency while the
|
||||||
|
* interface package owns the real schemas.
|
||||||
|
*/
|
||||||
|
export interface ProjectionSchemaLike<T> {
|
||||||
|
/**
|
||||||
|
* 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<K extends keyof SessionProjectionMap & string> {
|
||||||
|
key: K
|
||||||
|
/** Validates the baseline payload at the wire boundary (a failed parse degrades to capability absent). */
|
||||||
|
schema: ProjectionSchemaLike<SessionProjectionMap[K]>
|
||||||
|
/**
|
||||||
|
* 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 = {
|
||||||
|
<K extends keyof SessionProjectionMap & string>(key: K): SessionProjectionMap[K] | undefined
|
||||||
|
<K extends keyof SessionProjectionMap & string, S>(
|
||||||
|
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<string, unknown>
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Type-erased spec view the framework machinery works with (the register seam already proved the typed contract). */
|
||||||
|
interface ErasedCellSpec {
|
||||||
|
key: string
|
||||||
|
schema: ProjectionSchemaLike<unknown>
|
||||||
|
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<unknown> {
|
||||||
|
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<string, ProjectionCell>()
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 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<K extends keyof SessionProjectionMap & string>(spec: ProjectionCellSpec<K>): () => 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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -26,6 +26,7 @@ import { createScope, scopeOf as scopeTagOf } from '../agents/scope.ts'
|
|||||||
import { SessionManager } from './manager.ts'
|
import { SessionManager } from './manager.ts'
|
||||||
import type { SessionListPhase } from './manager.ts'
|
import type { SessionListPhase } from './manager.ts'
|
||||||
import type { Session } from './session.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. */
|
/** Session list row projected from the host list RPC plus live stream increments. */
|
||||||
export interface SessionSummary {
|
export interface SessionSummary {
|
||||||
@@ -165,6 +166,15 @@ export class SessionsService {
|
|||||||
private readonly scopes = new Map<SessionId, ScopeRecord>()
|
private readonly scopes = new Map<SessionId, ScopeRecord>()
|
||||||
/** Registered per-session standard-props providers, in registration order. */
|
/** Registered per-session standard-props providers, in registration order. */
|
||||||
private readonly providers: SessionProvideDescriptor[] = []
|
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<ProjectionCellSpec<keyof SessionProjectionMap & string>, Map<SessionId, () => void>>()
|
||||||
/** Static no-session projection, rebuilt only when the provider roster changes. */
|
/** Static no-session projection, rebuilt only when the provider roster changes. */
|
||||||
private maybeInfo: SessionMaybeProvideInfo
|
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<K extends keyof SessionProjectionMap & string>(spec: ProjectionCellSpec<K>): () => void {
|
||||||
|
const erased = spec as ProjectionCellSpec<keyof SessionProjectionMap & string>
|
||||||
|
const disposers = new Map<SessionId, () => 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. */
|
/** Rebuild every live scope's standard-props bundle after a provider roster change. */
|
||||||
private rematerializeProvideBundles(): void {
|
private rematerializeProvideBundles(): void {
|
||||||
this.maybeInfo = this.materializeMaybeProvideInfo()
|
this.maybeInfo = this.materializeMaybeProvideInfo()
|
||||||
@@ -254,7 +287,7 @@ export class SessionsService {
|
|||||||
props[name] = undefined
|
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). */
|
/** 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]
|
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);
|
// 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.
|
// mint and bind are one step so a live scope record implies a bound actx.
|
||||||
session.bindScope(ctx)
|
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 binding: SessionBinding = { sessionId: id, session, ctx }
|
||||||
const record: ScopeRecord = {
|
const record: ScopeRecord = {
|
||||||
fiber,
|
fiber,
|
||||||
@@ -559,6 +605,12 @@ export class SessionsService {
|
|||||||
// Release the Session's dispatch point with the scope it belongs to (a
|
// Release the Session's dispatch point with the scope it belongs to (a
|
||||||
// surviving instance — the live Intent — rebinds when resolve re-mints).
|
// surviving instance — the live Intent — rebinds when resolve re-mints).
|
||||||
record.binding.session.unbindScope()
|
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
|
// Optional lookup: slots and sessions are sibling services with no
|
||||||
// declared dependency; a slots-less boot (object-layer tests) skips.
|
// declared dependency; a slots-less boot (object-layer tests) skips.
|
||||||
this.rootCtx.get('slots')?.pruneStoreScope(id)
|
this.rootCtx.get('slots')?.pruneStoreScope(id)
|
||||||
|
|||||||
@@ -20,6 +20,8 @@ import { PendingWait } from './pending.ts'
|
|||||||
import { FoldAdapter } from './fold-adapter.ts'
|
import { FoldAdapter } from './fold-adapter.ts'
|
||||||
import { Notifier } from './notifier.ts'
|
import { Notifier } from './notifier.ts'
|
||||||
import { PartialAccumulator } from './partial.ts'
|
import { PartialAccumulator } from './partial.ts'
|
||||||
|
import { ProjectionCellSet } from './projection-cell.ts'
|
||||||
|
import type { ProjectionsBaseline } from './projection-cell.ts'
|
||||||
|
|
||||||
/** Messages requested per history page. */
|
/** Messages requested per history page. */
|
||||||
export const PAGE_MESSAGES = 50
|
export const PAGE_MESSAGES = 50
|
||||||
@@ -126,6 +128,17 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
|
|||||||
/** subscribed.lastSeq baseline (gap detection; null when no subscribed frame arrived — degrade to the liveBuffer dedup path). */
|
/** subscribed.lastSeq baseline (gap detection; null when no subscribed frame arrived — degrade to the liveBuffer dedup path). */
|
||||||
private subscribedLastSeq: number | null = null
|
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 snapshotCache: ConversationSnapshot
|
||||||
private readonly notifier = new Notifier(() => {
|
private readonly notifier = new Notifier(() => {
|
||||||
this.snapshotCache = this.buildSnapshot()
|
this.snapshotCache = this.buildSnapshot()
|
||||||
@@ -482,13 +495,13 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
|
|||||||
this.openError = result.error
|
this.openError = result.error
|
||||||
return
|
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.
|
// 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()
|
const tailSeq = this.windowTailSeq()
|
||||||
if (this.subscribedLastSeq !== null && tailSeq !== null && this.subscribedLastSeq > tailSeq) {
|
if (this.subscribedLastSeq !== null && tailSeq !== null && this.subscribedLastSeq > tailSeq) {
|
||||||
result = (await this.api.sessions.history({ sessionId: this.sessionId, maxMessages: PAGE_MESSAGES })).result
|
result = (await this.api.sessions.history({ sessionId: this.sessionId, maxMessages: PAGE_MESSAGES })).result
|
||||||
if (generation !== this.openGeneration) return
|
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'
|
this.openState = 'open'
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
@@ -505,8 +518,12 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
|
|||||||
/** Install the history window + stitch the liveBuffer (seq is the sole dedup key).
|
/** Install the history window + stitch the liveBuffer (seq is the sole dedup key).
|
||||||
* Stitching MUST NOT route through acceptLiveEvent: openState is still 'loading' here
|
* Stitching MUST NOT route through acceptLiveEvent: openState is still 'loading' here
|
||||||
* (doOpen flips it after install), so recursing would push every buffered event straight
|
* (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). */
|
* 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 {
|
* 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.events = entries.map(e => e.event)
|
||||||
this.views = entries.map(e => e.view)
|
this.views = entries.map(e => e.view)
|
||||||
this.baseSeq = this.events[0]?.seq ?? 0
|
this.baseSeq = this.events[0]?.seq ?? 0
|
||||||
@@ -521,6 +538,8 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
|
|||||||
this.todos = todos ?? []
|
this.todos = todos ?? []
|
||||||
this.foldAdapter.reset(this.events, this.baseSeq, this.views)
|
this.foldAdapter.reset(this.events, this.baseSeq, this.views)
|
||||||
this.rebuildDerivedFromWindow()
|
this.rebuildDerivedFromWindow()
|
||||||
|
if (projections !== undefined) this.projections.resetBaseline(projections)
|
||||||
|
this.projections.offerWindow(this.events)
|
||||||
const buffered = this.liveBuffer
|
const buffered = this.liveBuffer
|
||||||
this.liveBuffer = []
|
this.liveBuffer = []
|
||||||
for (const item of buffered) this.appendLive(item.event, item.view)
|
for (const item of buffered) this.appendLive(item.event, item.view)
|
||||||
@@ -535,6 +554,8 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
|
|||||||
this.views.push(view)
|
this.views.push(view)
|
||||||
this.foldAdapter.append(event, view)
|
this.foldAdapter.append(event, view)
|
||||||
this.applyEventSideEffects(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;
|
/** Land a live session/event (open/repair in flight -> buffer; overlapping seq -> drop;
|
||||||
@@ -569,7 +590,7 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
|
|||||||
const { result } = await this.api.sessions.history({ sessionId: this.sessionId, maxMessages: PAGE_MESSAGES })
|
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.
|
// 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') {
|
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) {
|
} catch (error) {
|
||||||
console.error('[web-runtime] gap repair failed:', error)
|
console.error('[web-runtime] gap repair failed:', error)
|
||||||
@@ -829,3 +850,19 @@ function derivePhase(hasContent: boolean, promptAttempted: boolean): ComposerPha
|
|||||||
if (hasContent) return 'active'
|
if (hasContent) return 'active'
|
||||||
return promptAttempted ? 'engaging' : 'blank'
|
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
|
||||||
|
}
|
||||||
|
|||||||
240
packages/client/runtime/tests/projection-cell.spec.ts
Normal file
240
packages/client/runtime/tests/projection-cell.spec.ts
Normal file
@@ -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()
|
||||||
|
})
|
||||||
|
})
|
||||||
Reference in New Issue
Block a user