diff --git a/packages/host/apiproxy/README.md b/packages/host/apiproxy/README.md index e450f70819..d23f64a880 100644 --- a/packages/host/apiproxy/README.md +++ b/packages/host/apiproxy/README.md @@ -10,7 +10,7 @@ Wire messages form a four-quadrant discriminated union — who initiates × requ The layering/protocol decisions are recorded in the [GUI layering and RPC protocol RFC](../../../.agents/notes/implemented/architecture/2026-07-19-gui-layering-and-rpc-protocol.md); the browser-side consumption architecture in the [web client architecture RFC](../../../.agents/notes/implemented/architecture/2026-07-19-gui-web-client-architecture.md). -`session.history`'s tail page (`beforeSeq` absent) additionally carries an optional `projections` block — one synchronous cut over every provider registered on `ctx.sessionProjections` (`@deepseek-ai/dsh-session-projection`), with `asOfSeq` equal to the window tail seq. The handler holds zero domain knowledge (each value passes its provider's own schema; the wire schema keeps `values` wide); loadOlder pages never carry the block, and a composition without the registry serves histories without it. +`session.history`'s tail page (`beforeSeq` absent) additionally carries an optional `projections` block — the watermark snapshot of every unit registered on `ctx.sessionProjections` (`@deepseek-ai/dsh-session-projection`), with `asOfSeq` = the last event seq the values reflect (`-1` on an empty log). The gateway also subscribes to the registry's change feed and mints a `session/projection` mux frame per changed unit (`{sessionId, key, value, seq}` — live push state, never logged; clients hold one generic per-session value store under higher-seq-wins). The carrier holds zero domain knowledge (each value passed its unit's own schema inside the registry; the wire schemas keep `values`/`value` wide); loadOlder pages never carry the block, and a composition without the registry serves histories without either surface. The mux stream projects the latest log-backed title as a validated `session/title` control frame after each attached-session subscription baseline and immediately after the corresponding live raw title event. This projection does not add titles to `session.list`; cold sessions remain metadata-only there until opening or resuming attaches their logs. diff --git a/packages/host/apiproxy/src/api-proxy.ts b/packages/host/apiproxy/src/api-proxy.ts index 92ce284dd4..110b94362a 100644 --- a/packages/host/apiproxy/src/api-proxy.ts +++ b/packages/host/apiproxy/src/api-proxy.ts @@ -299,25 +299,18 @@ function backscanTodos(events: readonly SessionEvent[]): TodoItem[] | undefined } /** - * Compute the projection baseline for one history tail page: read the - * session's next-event seq, then walk every registered provider — one fully - * synchronous pass (no await anywhere), so all values and `asOfSeq` form a - * single consistent cut and `asOfSeq` equals the window tail seq. Each value - * passes through its provider's own schema before leaving the host (the - * carrier holds zero domain knowledge; a provider returning an invalid value — - * including an accidental Promise from a non-synchronous `get` — fails loud - * here). An absent registry means the deployment has no projection seam: the - * whole block is absent and clients treat every key as capability-absent. + * The projection baseline for one history tail page: the registry's + * watermark-cache snapshot — one fully synchronous read (no await between the + * page slice and this), so all values and `asOfSeq` form a single consistent + * cut and `asOfSeq` equals the window tail event seq. The carrier holds zero + * domain knowledge (each value passed its unit's own schema inside the + * registry). An absent registry means the deployment has no projection seam: + * the whole block is absent and clients treat every key as capability-absent. */ function projectionsFor(ctx: Context, agent: Agent): SessionProjectionsBlock | undefined { const registry = ctx.get('sessionProjections') if (registry === undefined) return undefined - const asOfSeq = agent.session.seq - const values: Record = {} - for (const provider of registry.entries()) { - values[provider.key] = provider.schema.parse(provider.get(agent)) - } - return { asOfSeq, values: values as SessionProjectionsBlock['values'] } + return registry.snapshot(agent.session) } /** @@ -400,6 +393,16 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro for (const queue of muxQueues) queue.push(envelope) } + // Projection change feed → session/projection push frames. The carrier + // mints the wire frame (the seam package holds no wire vocabulary); the + // child activates only when a projection registry is composed, and the + // subscription unwinds with this gateway's fiber. + ctx.inject(['sessionProjections'], (projectionCtx) => { + projectionCtx.sessionProjections.onChanged((session, key, value, seq) => { + broadcast({ type: 'session/projection', sessionId: session.id, key, value, seq }) + }) + }) + /** * Per-session inbox mirror serving the mux-open queue snapshot (the same * refresh-recovery baseline as pending questions). Keyed by the stable diff --git a/packages/host/apiproxy/src/api/events.schema.ts b/packages/host/apiproxy/src/api/events.schema.ts index 973db5a91e..982e45dfe7 100644 --- a/packages/host/apiproxy/src/api/events.schema.ts +++ b/packages/host/apiproxy/src/api/events.schema.ts @@ -37,6 +37,9 @@ export const muxFrameSchema = z.discriminatedUnion('type', [ z.object({ type: z.literal('question/resolved'), sessionId: sessionIdSchema, questionRpcId: rpcIdSchema, outcome: z.union([z.literal('answered'), z.literal('cancelled')]) }), // content/source reuse the wide passthroughs (both are merge-extensible in core). z.object({ type: z.literal('session/queued'), sessionId: sessionIdSchema, content: z.array(contentBlockSchema), source: z.looseObject({ kind: z.string() }), steering: z.boolean() }), + // value stays wide: it already passed its unit's own schema on the host, + // and deep-validating here would import every domain's schema into the carrier. + z.object({ type: z.literal('session/projection'), sessionId: sessionIdSchema, key: z.string().min(1), value: z.unknown(), seq: z.number().int().nonnegative() }), z.object({ type: z.literal('stream/error'), error: rpcErrorSchema }), ]) as unknown as z.ZodType diff --git a/packages/host/apiproxy/src/api/events.ts b/packages/host/apiproxy/src/api/events.ts index 70139d0a00..28df8eb333 100644 --- a/packages/host/apiproxy/src/api/events.ts +++ b/packages/host/apiproxy/src/api/events.ts @@ -75,6 +75,15 @@ export type MuxFrame = * reconciliation key). */ | { type: 'session/queued'; sessionId: SessionId; content: ContentBlock[]; source: MessageSource; steering: boolean } + /** + * One projection unit's finished value changed (session-projection RFC). + * Live push state, never logged — replay recomputes on the host (the + * tool-view posture). `value` is the unit's schema-validated view output; + * `seq` is the unit's watermark at emission. Clients keep one generic + * per-session value store under higher-seq-wins, seeded by the history + * tail page's projections block. + */ + | { type: 'session/projection'; sessionId: SessionId; key: string; value: unknown; seq: number } | { type: 'stream/error'; error: RpcError } /** diff --git a/packages/host/apiproxy/src/api/sessions.schema.ts b/packages/host/apiproxy/src/api/sessions.schema.ts index f06231eaff..88ca7a9c96 100644 --- a/packages/host/apiproxy/src/api/sessions.schema.ts +++ b/packages/host/apiproxy/src/api/sessions.schema.ts @@ -105,7 +105,8 @@ export const todoItemSchema = z.object({ * deep-validating here would import every domain's schema into the carrier. */ export const sessionProjectionsBlockSchema = z.object({ - asOfSeq: z.number().int().nonnegative(), + // -1 = empty log (the lastSeq convention of session/subscribed). + asOfSeq: z.number().int().min(-1), values: z.record(z.string(), z.unknown()), }) as unknown as z.ZodType diff --git a/packages/host/apiproxy/src/api/sessions.ts b/packages/host/apiproxy/src/api/sessions.ts index 5579e638ee..eeacd8dd53 100644 --- a/packages/host/apiproxy/src/api/sessions.ts +++ b/packages/host/apiproxy/src/api/sessions.ts @@ -37,13 +37,16 @@ export interface HistoryEntry { /** * The projection baseline riding the history tail page: one synchronous cut - * over every registered projection provider. `asOfSeq` equals the window tail - * seq (the session's next-event seq at slice time) because the handler reads - * it and every value with no await in between. A key absent from `values` - * means the capability is absent (its domain plugin is unmounted). + * over every registered projection unit, read from the registry's watermark + * cache. `asOfSeq` is the seq of the last committed event every value + * reflects — the window tail event seq (`-1` for an empty log, mirroring + * `session/subscribed.lastSeq`), directly comparable with + * `session/projection` frame seqs under the client's higher-seq-wins rule. A + * key absent from `values` means the capability is absent (its domain plugin + * is unmounted). */ export interface SessionProjectionsBlock { - /** The session seq the values are consistent with (window tail seq). */ + /** Seq of the last event the values reflect; -1 for an empty log. */ asOfSeq: number /** Whole current value per registered projection key. */ values: Partial diff --git a/packages/host/apiproxy/tests/api-proxy-projections.spec.ts b/packages/host/apiproxy/tests/api-proxy-projections.spec.ts index 528fcd34b7..0da1610981 100644 --- a/packages/host/apiproxy/tests/api-proxy-projections.spec.ts +++ b/packages/host/apiproxy/tests/api-proxy-projections.spec.ts @@ -1,10 +1,10 @@ /** - * Projections block on the session.history tail page: a registered fake - * provider's whole value rides the tail page with asOfSeq equal to the window - * tail seq; loadOlder pages (beforeSeq present) never carry the block; a - * composition without the registry serves histories without the block; a - * disposed registration's key leaves subsequent responses; and a provider - * value rejected by its own schema fails the handler loud. + * Projection carrier paths of the host ApiProxy: the history tail page's + * projections block reads the registry's watermark snapshot (asOfSeq = last + * event seq, one consistent cut); loadOlder pages never carry the block; a + * composition without the registry serves histories without it; a disposed + * registration's key leaves subsequent responses; and every unit change is + * pushed to mux consumers as a session/projection frame minted here. */ import { describe, expect, it } from 'vitest' @@ -15,15 +15,15 @@ import type { Agent } from '@deepseek-ai/dsh-agent' import SessionStore from '@deepseek-ai/dsh-session' import type { Session } from '@deepseek-ai/dsh-session' import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection' -import type { ProjectionProvider } from '@deepseek-ai/dsh-session-projection' +import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection' import UserInteractionService from '@deepseek-ai/dsh-user-interaction' -import type { RpcRequest } from '@deepseek-ai/dsh-host-apiproxy/api/rpc' +import type { MuxFrame, RpcRequest } from '@deepseek-ai/dsh-host-apiproxy/api' import { RpcId } from '@deepseek-ai/dsh-host-apiproxy/api/rpc' import { createApiProxy } from '@deepseek-ai/dsh-host-apiproxy' -declare module '@deepseek-ai/dsh-session-projection' { +declare module '@deepseek-ai/dsh-session-projection/types' { interface SessionProjectionMap { - 'test/echo-seq': { seenSeq: number } + 'test/last-user': { text: string } | null } } @@ -32,12 +32,18 @@ function request

(payload: P): RpcRequest

{ return { rpcId: RpcId(`proj-${String(nextRpc++)}`), payload } } -/** Provider whose value records the session seq it observed at get() time. */ -const echoSeqProvider: ProjectionProvider<'test/echo-seq'> = { - key: 'test/echo-seq', - schema: z.object({ seenSeq: z.number().int().nonnegative() }), - get: agent => ({ seenSeq: agent.session.seq }), -} +/** Whole-value unit folding the latest user/message text; null before the first. */ +type LastUserState = { text: string } | null +const lastUserUnit = (): ProjectionDefinition<'test/last-user', LastUserState> => ({ + key: 'test/last-user', + schema: z.union([z.object({ text: z.string() }), z.null()]), + init: () => null, + apply: (state, event) => (event.type === 'user/message' + ? { text: (event.data.content[0] as { text?: string }).text ?? '' } + : state), + view: state => state, + stateVersion: 1, +}) async function harness(withRegistry: boolean): Promise<{ ctx: Context; session: Session }> { const ctx = new Context() @@ -59,32 +65,29 @@ function seedMessages(session: Session, count: number): void { } } -describe('session.history projections block', () => { - it('serves the registered value on the tail page with asOfSeq = window tail seq', async () => { - const { ctx, session } = await harness(true) - ctx.sessionProjections.register(echoSeqProvider) - seedMessages(session, 3) - const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' }) +const api = (ctx: Context) => createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' }) - const response = await api.sessions.history(request({ sessionId: session.id })) +describe('session.history projections block', () => { + it('serves the unit value on the tail page with asOfSeq = last event seq', async () => { + const { ctx, session } = await harness(true) + ctx.sessionProjections.register(lastUserUnit()) + seedMessages(session, 3) + const response = await api(ctx).sessions.history(request({ sessionId: session.id })) expect(response.result.ok).toBe(true) if (!response.result.ok) throw new Error('unreachable') const { events, projections } = response.result.value expect(projections).toBeDefined() - expect(projections?.asOfSeq).toBe(session.seq) - // The cut is consistent: the value observed the same seq the block stamps. - expect(projections?.values['test/echo-seq']).toEqual({ seenSeq: session.seq }) - // asOfSeq is the window tail: the last served event sits right below it. - expect(events.at(-1)?.event.seq).toBe(session.seq - 1) + expect(projections?.asOfSeq).toBe(session.seq - 1) + expect(projections?.values['test/last-user']).toEqual({ text: 'm2' }) + // asOfSeq IS the window tail: the last served event carries it. + expect(events.at(-1)?.event.seq).toBe(projections?.asOfSeq) }) it('never carries the block on loadOlder pages (beforeSeq present)', async () => { const { ctx, session } = await harness(true) - ctx.sessionProjections.register(echoSeqProvider) + ctx.sessionProjections.register(lastUserUnit()) seedMessages(session, 5) - const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' }) - - const older = await api.sessions.history(request({ sessionId: session.id, beforeSeq: 3, maxMessages: 2 })) + const older = await api(ctx).sessions.history(request({ sessionId: session.id, beforeSeq: 3, maxMessages: 2 })) expect(older.result.ok).toBe(true) if (!older.result.ok) throw new Error('unreachable') expect('projections' in older.result.value).toBe(false) @@ -93,9 +96,7 @@ describe('session.history projections block', () => { it('serves no block when the composition has no projection registry', async () => { const { ctx, session } = await harness(false) seedMessages(session, 2) - const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' }) - - const response = await api.sessions.history(request({ sessionId: session.id })) + const response = await api(ctx).sessions.history(request({ sessionId: session.id })) expect(response.result.ok).toBe(true) if (!response.result.ok) throw new Error('unreachable') expect('projections' in response.result.value).toBe(false) @@ -103,35 +104,78 @@ describe('session.history projections block', () => { it('drops a disposed registration from subsequent tail pages (empty block, key absent)', async () => { const { ctx, session } = await harness(true) - const dispose = ctx.sessionProjections.register(echoSeqProvider) + const dispose = ctx.sessionProjections.register(lastUserUnit()) seedMessages(session, 1) - const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' }) - - const before = await api.sessions.history(request({ sessionId: session.id })) + const proxy = api(ctx) + const before = await proxy.sessions.history(request({ sessionId: session.id })) if (!before.result.ok) throw new Error('unreachable') - expect(before.result.value.projections?.values['test/echo-seq']).toBeDefined() + expect(before.result.value.projections?.values['test/last-user']).toEqual({ text: 'm0' }) dispose() - const after = await api.sessions.history(request({ sessionId: session.id })) + const after = await proxy.sessions.history(request({ sessionId: session.id })) if (!after.result.ok) throw new Error('unreachable') // The registry is still mounted, so the block itself stays (asOfSeq cut // with zero keys); the disposed key reads as capability absence. - expect(after.result.value.projections?.asOfSeq).toBe(session.seq) + expect(after.result.value.projections?.asOfSeq).toBe(session.seq - 1) expect(after.result.value.projections?.values).toEqual({}) }) +}) - it('fails loud when a provider value violates its own schema (async get is unrepresentable)', async () => { +describe('session/projection push frame', () => { + /** Drain frames until `count` session/projection frames arrived. */ + async function collect(iterable: AsyncIterable>, count: number, abort: AbortController): Promise { + const frames: MuxFrame[] = [] + for await (const envelope of iterable) { + frames.push(envelope.payload) + if (frames.filter(f => f.type === 'session/projection').length >= count) abort.abort() + } + return frames + } + + it('broadcasts a frame per changed unit with the causing seq, and none for same-reference applies', async () => { const { ctx, session } = await harness(true) - ctx.sessionProjections.register({ - key: 'test/echo-seq', - schema: z.object({ seenSeq: z.number().int().nonnegative() }), - // A Promise (what an accidentally-async get would return) is not the - // declared shape: the boundary parse rejects it before it hits the wire. - get: () => Promise.resolve({ seenSeq: 0 }) as never, - }) - seedMessages(session, 1) - const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' }) + ctx.sessionProjections.register(lastUserUnit()) + const proxy = api(ctx) + // The gateway's onChanged subscription lives in an inject child whose + // fiber activates asynchronously; yield until it lands before appending. + await new Promise(resolve => setTimeout(resolve, 0)) + const abort = new AbortController() + const stream = proxy.events.mux({ rpcId: RpcId('t-proj-mux'), payload: {} }, abort.signal) + const collected = collect(stream, 2, abort) - await expect(api.sessions.history(request({ sessionId: session.id }))).rejects.toThrow() + seedMessages(session, 1) + // Same-reference apply: turn/start does not concern the unit — no frame. + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + seedMessages(session, 1) + + const frames = await collected + const pushes = frames.filter( + (f): f is Extract => f.type === 'session/projection', + ) + expect(pushes).toEqual([ + { type: 'session/projection', sessionId: session.id, key: 'test/last-user', value: { text: 'm0' }, seq: 0 }, + { type: 'session/projection', sessionId: session.id, key: 'test/last-user', value: { text: 'm0' }, seq: 2 }, + ]) + // Frame seq aligns with the tail block's asOfSeq vocabulary (higher-seq-wins compatible). + const tail = await proxy.sessions.history(request({ sessionId: session.id })) + if (!tail.result.ok) throw new Error('unreachable') + expect(tail.result.value.projections?.asOfSeq).toBe(pushes.at(-1)?.seq) + }) + + it('emits no projection frames when the composition has no registry', async () => { + const { ctx, session } = await harness(false) + const proxy = api(ctx) + const abort = new AbortController() + const stream = proxy.events.mux({ rpcId: RpcId('t-noproj-mux'), payload: {} }, abort.signal) + const frames: MuxFrame[] = [] + const drained = (async () => { + for await (const envelope of stream) { + frames.push(envelope.payload) + if (frames.filter(f => f.type === 'session/event').length >= 2) abort.abort() + } + })() + seedMessages(session, 2) + await drained + expect(frames.some(f => f.type === 'session/projection')).toBe(false) }) })