diff --git a/packages/host/apiproxy/README.md b/packages/host/apiproxy/README.md index 253c0974cc..69e616193b 100644 --- a/packages/host/apiproxy/README.md +++ b/packages/host/apiproxy/README.md @@ -10,6 +10,8 @@ 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. + 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. Workspace and Session lists are separate reconnect baselines. `workspace.create` creates a unique name or adopts an existing directory, `workspace.delete` removes only the Workspace registration, `session.create` accepts an optional preallocated Session id, and `host/workspace-changed`, `host/workspace-removed`, plus `host/session-added` carry committed increments in either arrival order. Registration deletion preserves the directory and session logs; its Sessions remain in `session.list` and become Ungrouped. `SessionSummary.blank` and the `host/session-added` frame carry the derived zero-events bit: clients hide blank sessions and reuse them per workspace, flip blank on the first `host/session-status(running:true)`, and treat `session.list` as the reconnect authority; cold summaries are never blank because lazy persistence keeps never-appended sessions out of `list()`. diff --git a/packages/host/apiproxy/package.json b/packages/host/apiproxy/package.json index 0c7107a1f9..de2263063f 100644 --- a/packages/host/apiproxy/package.json +++ b/packages/host/apiproxy/package.json @@ -46,6 +46,7 @@ "@deepseek-ai/dsh-llm": "workspace:^", "@deepseek-ai/dsh-session": "workspace:^", "@deepseek-ai/dsh-session-persistence": "workspace:^", + "@deepseek-ai/dsh-session-projection": "workspace:^", "@deepseek-ai/dsh-session-title": "workspace:^", "@deepseek-ai/dsh-skill": "workspace:^", "@deepseek-ai/dsh-tools": "workspace:^", diff --git a/packages/host/apiproxy/src/api-proxy.ts b/packages/host/apiproxy/src/api-proxy.ts index 5cd1a6dc4c..ae23d9fd47 100644 --- a/packages/host/apiproxy/src/api-proxy.ts +++ b/packages/host/apiproxy/src/api-proxy.ts @@ -21,9 +21,11 @@ import { // Type-only: brings the `ctx.tools` Context merge into this program (viewFor reads presenters). import type {} from '@deepseek-ai/dsh-tools' import type { - ApiProxy, HistoryEntry, HostFrame, MuxFrame, QuestionResponsePayload, SessionSummary, ToolEventView, - WorkspaceId, WorkspaceView, + ApiProxy, HistoryEntry, HostFrame, MuxFrame, QuestionResponsePayload, SessionProjectionsBlock, + SessionSummary, ToolEventView, WorkspaceId, WorkspaceView, } from './api/index.ts' +// Type-only: resolves `ctx.get('sessionProjections')` to the projection registry. +import type {} from '@deepseek-ai/dsh-session-projection' // Type-only edges: resolve `ctx.get('commands')`, the `commands/change` event, and `ctx.get('skills')`. import type {} from '@deepseek-ai/dsh-commands' import type {} from '@deepseek-ai/dsh-skill' @@ -296,6 +298,28 @@ function backscanTodos(events: readonly SessionEvent[]): TodoItem[] | undefined return 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. + */ +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'] } +} + /** * Thrown by the cold-resume path when the id names no servable session * (absent from the store, or a pre-project legacy log without a cwd). @@ -657,6 +681,8 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro const { sessionId, beforeSeq, maxMessages } = request.payload const found = await agentFor(sessionId) if ('error' in found) return err(request, found.error) + // Everything below the resume above is synchronous: the page slice, + // the seq read, and the projection walk see one un-torn session state. const page = paginate(found.agent.session.events, beforeSeq, maxMessages ?? DEFAULT_MAX_MESSAGES) // Views are computed against the registry at pagination time; result // pairing scans within the page only (message-boundary pagination keeps @@ -668,8 +694,17 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro // Tail page carries the session-level todo projection over the FULL // log (the page window may not contain the last todo/write; a paged // client cannot reconstruct session-level state from it). + // TODO(gui): retire this rider onto the generic projections block. const todos = beforeSeq === undefined ? backscanTodos(found.agent.session.events) : undefined - return ok(request, { events: entries, hasMore: page.hasMore, ...todos === undefined ? {} : { todos } }) + // Baseline rider: tail page only — loadOlder (beforeSeq present) is + // the one path that never needs a fresh projection baseline. + const projections = beforeSeq === undefined ? projectionsFor(ctx, found.agent) : undefined + return ok(request, { + events: entries, + hasMore: page.hasMore, + ...todos === undefined ? {} : { todos }, + ...projections === undefined ? {} : { projections }, + }) }, async prompt(request) { diff --git a/packages/host/apiproxy/src/api/index.ts b/packages/host/apiproxy/src/api/index.ts index 537b2744ef..23b08a2ef0 100644 --- a/packages/host/apiproxy/src/api/index.ts +++ b/packages/host/apiproxy/src/api/index.ts @@ -25,7 +25,7 @@ export interface ApiProxy { } // ---- Domain interfaces and payload entities ---- -export type { HistoryEntry, SessionsApi, SessionSummary } from './sessions.ts' +export type { HistoryEntry, SessionProjectionsBlock, SessionsApi, SessionSummary } from './sessions.ts' export type { HostApi } from './host.ts' export type { WorkspaceApi, WorkspaceId, WorkspaceView } from './workspace.ts' export type { CommandsApi, CommandDescriptor, CommandExecuteResult } from './commands.ts' diff --git a/packages/host/apiproxy/src/api/sessions.schema.ts b/packages/host/apiproxy/src/api/sessions.schema.ts index 9445568e98..f06231eaff 100644 --- a/packages/host/apiproxy/src/api/sessions.schema.ts +++ b/packages/host/apiproxy/src/api/sessions.schema.ts @@ -9,7 +9,7 @@ import { z } from 'zod' import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session/types' import type { RequestPayload, ResponseValue } from './rpc-map.ts' import type { Wire } from './rpc.schema.ts' -import type { HistoryEntry, SessionSummary } from './sessions.ts' +import type { HistoryEntry, SessionProjectionsBlock, SessionSummary } from './sessions.ts' import type { ToolEventView } from './events.ts' import type { WorkspaceId } from './workspace.ts' @@ -99,11 +99,22 @@ export const todoItemSchema = z.object({ status: z.union([z.literal('pending'), z.literal('in_progress'), z.literal('completed')]), }) -/** session.history response value. */ +/** + * Projection baseline passthrough: `values` stays a wide record — each value + * was already parsed by its provider's own schema on the host side, and + * deep-validating here would import every domain's schema into the carrier. + */ +export const sessionProjectionsBlockSchema = z.object({ + asOfSeq: z.number().int().nonnegative(), + values: z.record(z.string(), z.unknown()), +}) as unknown as z.ZodType + +/** session.history response value (todos and projections ride the tail page only). */ export const sessionHistoryValueSchema = z.object({ events: z.array(historyEntrySchema), hasMore: z.boolean(), todos: z.array(todoItemSchema).optional(), + projections: sessionProjectionsBlockSchema.optional(), }) satisfies z.ZodType>> /** ContentBlock passthrough: core is merge-extensible — the type discriminant envelope is strict, the rest stays wide. */ diff --git a/packages/host/apiproxy/src/api/sessions.ts b/packages/host/apiproxy/src/api/sessions.ts index e46bc43fe8..00e2511862 100644 --- a/packages/host/apiproxy/src/api/sessions.ts +++ b/packages/host/apiproxy/src/api/sessions.ts @@ -6,6 +6,7 @@ import type { ContentBlock } from '@deepseek-ai/dsh-llm/types' import type { SessionEvent, SessionId, TodoItem } from '@deepseek-ai/dsh-session/types' +import type { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection' import type { RpcId, RpcRequest, RpcResponse } from './rpc.ts' import type { ToolEventView } from './events.ts' import type { WorkspaceId } from './workspace.ts' @@ -32,6 +33,20 @@ export interface HistoryEntry { view?: ToolEventView } +/** + * 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). + */ +export interface SessionProjectionsBlock { + /** The session seq the values are consistent with (window tail seq). */ + asOfSeq: number + /** Whole current value per registered projection key. */ + values: Partial +} + /** Session list entry (v1 builds no index: list does readdir+stat). */ export interface SessionSummary { sessionId: SessionId @@ -81,9 +96,15 @@ export interface SessionsApi { * projection (latest `todo/write` over the FULL log, independent of the page window) — * so a paged client restores the plan without walking history; absent when the session * never wrote one. Older pages omit it (the projection is session-level, not per-page). + * TODO(gui): the todos rider retires onto the generic projections block below. + * The tail page — and only the tail page — additionally carries `projections` + * when the deployment mounts the session-projection registry: every moment + * the client needs a fresh baseline already pulls the tail page, and + * loadOlder (the only beforeSeq path) is the only path that never needs one. + * A deployment without the registry serves histories without the block. */ history(request: RpcRequest<{ sessionId: SessionId; beforeSeq?: number; maxMessages?: number }>): - Promise> + Promise> /** Sends a message. content is core's ContentBlock[] verbatim; mode maps 1:1 — queue→send, steer→steer. */ prompt(request: RpcRequest<{ sessionId: SessionId; mode: 'queue' | 'steer'; content: ContentBlock[] }>): diff --git a/packages/host/apiproxy/tests/api-proxy-projections.spec.ts b/packages/host/apiproxy/tests/api-proxy-projections.spec.ts new file mode 100644 index 0000000000..528fcd34b7 --- /dev/null +++ b/packages/host/apiproxy/tests/api-proxy-projections.spec.ts @@ -0,0 +1,137 @@ +/** + * 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. + */ + +import { describe, expect, it } from 'vitest' +import { Context } from 'cordis' +import { z } from 'zod' +import AgentRegistry from '@deepseek-ai/dsh-agent' +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 UserInteractionService from '@deepseek-ai/dsh-user-interaction' +import type { RpcRequest } from '@deepseek-ai/dsh-host-apiproxy/api/rpc' +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' { + interface SessionProjectionMap { + 'test/echo-seq': { seenSeq: number } + } +} + +let nextRpc = 1 +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 }), +} + +async function harness(withRegistry: boolean): Promise<{ ctx: Context; session: Session }> { + const ctx = new Context() + await ctx.plugin(SessionStore) + await ctx.plugin(UserInteractionService) + await ctx.plugin(AgentRegistry) + if (withRegistry) await ctx.plugin(SessionProjectionRegistry) + const session = ctx.sessions.create() + // history resolves the agent first; a live structural stub is enough (only + // .session is read on this path). + ctx.agents.register({ id: session.id, session, status: 'idle', ctx } as Agent) + return { ctx, session } +} + +/** Append `count` user messages so the log has paginable message boundaries. */ +function seedMessages(session: Session, count: number): void { + for (let i = 0; i < count; i++) { + session.append('user/message', { content: [{ type: 'text', text: `m${i}` }], source: { kind: 'user' } }, { surfaceOp: 'append' }) + } +} + +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 response = await api.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) + }) + + it('never carries the block on loadOlder pages (beforeSeq present)', async () => { + const { ctx, session } = await harness(true) + ctx.sessionProjections.register(echoSeqProvider) + 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 })) + expect(older.result.ok).toBe(true) + if (!older.result.ok) throw new Error('unreachable') + expect('projections' in older.result.value).toBe(false) + }) + + 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 })) + expect(response.result.ok).toBe(true) + if (!response.result.ok) throw new Error('unreachable') + expect('projections' in response.result.value).toBe(false) + }) + + 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) + 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 })) + if (!before.result.ok) throw new Error('unreachable') + expect(before.result.value.projections?.values['test/echo-seq']).toBeDefined() + + dispose() + const after = await api.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?.values).toEqual({}) + }) + + it('fails loud when a provider value violates its own schema (async get is unrepresentable)', 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' }) + + await expect(api.sessions.history(request({ sessionId: session.id }))).rejects.toThrow() + }) +}) diff --git a/packages/host/apiproxy/tsconfig.json b/packages/host/apiproxy/tsconfig.json index f5aabb1cf8..4f5d52ed73 100644 --- a/packages/host/apiproxy/tsconfig.json +++ b/packages/host/apiproxy/tsconfig.json @@ -32,6 +32,9 @@ { "path": "../../session-persistence/session-persistence" }, + { + "path": "../../session-projection/session-projection" + }, { "path": "../../session-title/session-title" }, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7795770e40..5322fa9a7f 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2604,6 +2604,9 @@ importers: '@deepseek-ai/dsh-session-persistence': specifier: workspace:^ version: link:../../session-persistence/session-persistence + '@deepseek-ai/dsh-session-projection': + specifier: workspace:^ + version: link:../../session-projection/session-projection '@deepseek-ai/dsh-session-title': specifier: workspace:^ version: link:../../session-title/session-title