From 003b22a1572569a3cf77efde193ade8cfebc75a0 Mon Sep 17 00:00:00 2001 From: imccyu <276526105+imccyu@users.noreply.github.com> Date: Tue, 28 Jul 2026 01:38:15 +0800 Subject: [PATCH] =?UTF-8?q?feat(apiproxy):=20projection=20column=20on=20se?= =?UTF-8?q?ssion.list=20=E2=80=94=20cold=20titles=20with=20zero=20log=20lo?= =?UTF-8?q?ads?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SessionSummary grows an optional projections column (whole value per key, same passthrough posture as the history-tail block): attached rows cut the live registry watermark cache; cold rows view the persisted projection cache's stored rows via the new registry viewCheckpoint face (version- matching keys only, zero I/O) — the RFC's motivating scenario, every session's title across a listing without loading one event log. The column is fail-soft and absence-coded: no registry, no cache row, or a throwing read serve the row without the column, never breaking the listing. --- packages/host/apiproxy/package.json | 1 + packages/host/apiproxy/src/api-proxy.ts | 40 +++++++++- .../host/apiproxy/src/api/sessions.schema.ts | 11 +++ packages/host/apiproxy/src/api/sessions.ts | 11 +++ .../tests/api-proxy-projections.spec.ts | 73 ++++++++++++++++++- packages/host/apiproxy/tsconfig.json | 3 + .../session-projection-cache/src/index.ts | 16 +++- .../session-projection/src/index.ts | 21 ++++++ .../session-projection/tests/registry.spec.ts | 13 ++++ pnpm-lock.yaml | 3 + 10 files changed, 187 insertions(+), 5 deletions(-) diff --git a/packages/host/apiproxy/package.json b/packages/host/apiproxy/package.json index ab632199bd..a3e0cfac5d 100644 --- a/packages/host/apiproxy/package.json +++ b/packages/host/apiproxy/package.json @@ -47,6 +47,7 @@ "@deepseek-ai/dsh-session": "workspace:^", "@deepseek-ai/dsh-session-persistence": "workspace:^", "@deepseek-ai/dsh-session-projection": "workspace:^", + "@deepseek-ai/dsh-session-projection-cache": "workspace:^", "@deepseek-ai/dsh-skill": "workspace:^", "@deepseek-ai/dsh-tools": "workspace:^", "@deepseek-ai/dsh-user-approval": "workspace:^", diff --git a/packages/host/apiproxy/src/api-proxy.ts b/packages/host/apiproxy/src/api-proxy.ts index 56f488857f..486a528a78 100644 --- a/packages/host/apiproxy/src/api-proxy.ts +++ b/packages/host/apiproxy/src/api-proxy.ts @@ -29,7 +29,9 @@ import type { WorkspaceId, WorkspaceView, } from './api/index.ts' // Type-only: resolves `ctx.get('sessionProjections')` to the projection registry. -import type {} from '@deepseek-ai/dsh-session-projection' +import type { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection' +// Type-only: resolves `ctx.get('sessionProjectionCache')` (the cold listing column). +import type {} from '@deepseek-ai/dsh-session-projection-cache' // 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' @@ -297,6 +299,26 @@ function projectionsFor(ctx: Context, agent: Agent): SessionProjectionsBlock | u return registry.snapshot(agent.session) } +/** + * The projection column of one session.list row, fail-soft: attached + * sessions cut the registry's live watermark cache; cold sessions view the + * persisted projection cache's stored rows (zero log loads either way — the + * listing use case the cache exists for). Any failure — and an empty value + * set — yields an absent column: a listing without projections is degraded, + * never broken. + */ +function listProjectionsFor(ctx: Context, id: SessionId, session: Session | undefined): Partial | undefined { + try { + const values = session !== undefined + ? ctx.get('sessionProjections')?.snapshot(session).values + : ctx.get('sessionProjectionCache')?.cachedValues(id) + return values !== undefined && Object.keys(values).length > 0 ? values : undefined + } catch (error) { + ctx.logger.warn(`session.list: projection column for "${id}" failed (serving the row without it): ${String(error)}`) + return undefined + } +} + /** * 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). @@ -654,13 +676,25 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro async list(request) { const items = ctx.sessions.list().map((session) => { const agent = ctx.agents.get(session.id) - return summarize(session, agent?.status === 'running') + const projections = listProjectionsFor(ctx, session.id, session) + return { + ...summarize(session, agent?.status === 'running'), + ...projections === undefined ? {} : { projections }, + } }) const attached = new Set(items.map(item => item.sessionId)) const persistence = ctx.get('sessionPersistence') if (persistence !== undefined) { const cold = (await persistence.list()).filter(meta => !attached.has(meta.id) && meta.cwd !== undefined) - items.push(...await Promise.all(cold.map(meta => summarizeCold(persistence, meta)))) + items.push(...await Promise.all(cold.map(async (meta) => { + // Cold rows read the persisted projection cache only — never a + // log load; a session without a cache row simply has no column. + const projections = listProjectionsFor(ctx, meta.id, undefined) + return { + ...await summarizeCold(persistence, meta), + ...projections === undefined ? {} : { projections }, + } + }))) } items.sort((a, b) => b.updatedAt - a.updatedAt) return ok(request, { items }) diff --git a/packages/host/apiproxy/src/api/sessions.schema.ts b/packages/host/apiproxy/src/api/sessions.schema.ts index 5c674337bc..a02267a6bf 100644 --- a/packages/host/apiproxy/src/api/sessions.schema.ts +++ b/packages/host/apiproxy/src/api/sessions.schema.ts @@ -7,6 +7,7 @@ import { z } from 'zod' import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session/types' +import type { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection/types' import type { RequestPayload, ResponseValue } from './rpc-map.ts' import type { Wire } from './rpc.schema.ts' import type { @@ -37,6 +38,15 @@ export const sessionEventSchema = z.object({ surfaceOp: z.unknown().optional(), }) as unknown as z.ZodType +/** + * Projection-values passthrough (same posture as + * {@link sessionProjectionsBlockSchema}): each value already passed its + * unit's own schema on the host side; deep-validating here would import + * every domain's schema into the carrier. + */ +const projectionValuesSchema = + z.record(z.string(), z.unknown()) as unknown as z.ZodType> + /** SessionSummary row of session.list. */ export const sessionSummarySchema = z.object({ sessionId: sessionIdSchema, @@ -45,6 +55,7 @@ export const sessionSummarySchema = z.object({ blank: z.boolean(), parentSessionId: sessionIdSchema.optional(), cwd: z.string().optional(), + projections: projectionValuesSchema.optional(), }) satisfies z.ZodType> /** session.list request payload (cursor is a reserved seat, unimplemented in v1). */ diff --git a/packages/host/apiproxy/src/api/sessions.ts b/packages/host/apiproxy/src/api/sessions.ts index 8c7be54440..a6d5c1517c 100644 --- a/packages/host/apiproxy/src/api/sessions.ts +++ b/packages/host/apiproxy/src/api/sessions.ts @@ -143,6 +143,17 @@ export interface SessionSummary { parentSessionId?: SessionId /** Session working directory (header.cwd passthrough); absent when unrecorded. */ cwd?: string + /** + * Whole current value per projection key, with zero log loads: attached + * sessions read the registry's live watermark cut; cold sessions read the + * persisted projection cache's stored rows — as stale as that session's + * last durable checkpoint, never wrong, superseded by the history tail + * baseline the moment the session is opened. Absent when no value is + * available (no registry, no cache row for a cold session, or a fail-soft + * cache read miss); a listing client treats absence as "no title yet", + * exactly like a blank session. + */ + projections?: Partial } /** Session-domain unary methods (the map keys session.* of RpcMethodMap). */ diff --git a/packages/host/apiproxy/tests/api-proxy-projections.spec.ts b/packages/host/apiproxy/tests/api-proxy-projections.spec.ts index e8999bdeda..caa957d05d 100644 --- a/packages/host/apiproxy/tests/api-proxy-projections.spec.ts +++ b/packages/host/apiproxy/tests/api-proxy-projections.spec.ts @@ -13,7 +13,7 @@ import { z } from 'zod' import AgentRegistry from '@deepseek-ai/dsh-agent' import type { Agent } from '@deepseek-ai/dsh-agent' import { createUserMessage } from '@deepseek-ai/dsh-llm' -import SessionStore from '@deepseek-ai/dsh-session' +import SessionStore, { SessionId } from '@deepseek-ai/dsh-session' import type { Session } from '@deepseek-ai/dsh-session' import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection' import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection' @@ -125,6 +125,77 @@ describe('session.history projections block', () => { }) }) +describe('session.list projections column', () => { + it('serves attached rows from the live registry cut', async () => { + const { ctx, session } = await harness(true) + ctx.sessionProjections.register(lastUserUnit()) + seedMessages(session, 1) + const response = await api(ctx).sessions.list(request({})) + if (!response.result.ok) throw new Error('unreachable') + const row = response.result.value.items.find(item => item.sessionId === session.id) + expect(row?.projections?.['test/last-user']).toEqual({ text: 'm0' }) + }) + + it('omits the column entirely when no registry is mounted', async () => { + const { ctx, session } = await harness(false) + seedMessages(session, 1) + const response = await api(ctx).sessions.list(request({})) + if (!response.result.ok) throw new Error('unreachable') + const row = response.result.value.items.find(item => item.sessionId === session.id) + expect(row).toBeDefined() + expect(row !== undefined && 'projections' in row).toBe(false) + }) + + it('serves cold rows from the persisted projection cache with zero log loads', async () => { + const { ctx } = await harness(true) + const coldId = SessionId('session-cold-listing') + const load = () => { throw new Error('list must not load event logs') } + ctx.provide('sessionPersistence', { + list: async () => [{ version: 0, id: coldId, createdAt: 5, cwd: '/tmp' }], + locate: () => undefined, + load, + inspect: load, + readFrom: load, + } as never) + ctx.provide('sessionProjectionCache', { + cachedValues: (id: unknown) => (id === coldId ? { 'test/last-user': { text: 'cached' } } : {}), + } as never) + const response = await api(ctx).sessions.list(request({})) + if (!response.result.ok) throw new Error('unreachable') + const row = response.result.value.items.find(item => item.sessionId === coldId) + expect(row?.running).toBe(false) + expect(row?.projections?.['test/last-user']).toEqual({ text: 'cached' }) + }) + + it('cold rows without a cache plugin (or without a stored row) just lack the column', async () => { + const { ctx } = await harness(true) + const coldId = SessionId('session-cold-uncached') + ctx.provide('sessionPersistence', { + list: async () => [{ version: 0, id: coldId, createdAt: 5, cwd: '/tmp' }], + locate: () => undefined, + } as never) + const response = await api(ctx).sessions.list(request({})) + if (!response.result.ok) throw new Error('unreachable') + const row = response.result.value.items.find(item => item.sessionId === coldId) + expect(row).toBeDefined() + expect(row !== undefined && 'projections' in row).toBe(false) + }) + + it('a throwing column read degrades that row, never the listing', async () => { + const { ctx, session } = await harness(true) + ctx.sessionProjections.register({ + ...lastUserUnit(), + view: () => { throw new Error('unit exploded') }, + }) + seedMessages(session, 1) + const response = await api(ctx).sessions.list(request({})) + if (!response.result.ok) throw new Error('unreachable') + const row = response.result.value.items.find(item => item.sessionId === session.id) + expect(row).toBeDefined() + expect(row !== undefined && 'projections' in row).toBe(false) + }) +}) + describe('session/projection push frame', () => { /** Drain frames until `count` session/projection frames arrived. */ async function collect(iterable: AsyncIterable>, count: number, abort: AbortController): Promise { diff --git a/packages/host/apiproxy/tsconfig.json b/packages/host/apiproxy/tsconfig.json index bf65db029d..fbe8e77719 100644 --- a/packages/host/apiproxy/tsconfig.json +++ b/packages/host/apiproxy/tsconfig.json @@ -35,6 +35,9 @@ { "path": "../../session-projection/session-projection" }, + { + "path": "../../session-projection/session-projection-cache" + }, { "path": "../../skill/skill" }, diff --git a/packages/session-projection/session-projection-cache/src/index.ts b/packages/session-projection/session-projection-cache/src/index.ts index c9336dd848..abb39ad1de 100644 --- a/packages/session-projection/session-projection-cache/src/index.ts +++ b/packages/session-projection/session-projection-cache/src/index.ts @@ -19,7 +19,7 @@ import type { Session, SessionEvent, SessionId } from '@deepseek-ai/dsh-session' // Empty type import: applies the package's cordis Context merge // (`ctx.sessionPersistence`), which this service reads on the cold path. import type {} from '@deepseek-ai/dsh-session-persistence' -import type { ProjectionCheckpoint, ProjectionSnapshot } from '@deepseek-ai/dsh-session-projection' +import type { ProjectionCheckpoint, ProjectionSnapshot, SessionProjectionMap } from '@deepseek-ai/dsh-session-projection' import type { KvTable } from '@deepseek-ai/dsh-storage-domain' import { projectionCacheDomainSpec } from './spec.ts' import type { CheckpointRecord } from './spec.ts' @@ -98,6 +98,20 @@ export class SessionProjectionCache extends Service { return this.requireTable().get(id)?.rows ?? {} } + /** + * The zero-I/O listing read: whole values viewed straight from the stored + * rows (version-matching keys only), as stale as the last durable + * checkpoint but never wrong. Synchronous — a listing over every stored + * session touches no log. Fresher paths (the history tail baseline, + * {@link coldSnapshot}) supersede these values whenever a session is + * actually opened. + * @param id - the session whose cached values are viewed. + * @returns whole values per key with a usable row; empty when none stored. + */ + cachedValues(id: SessionId): Partial { + return this.ctx.sessionProjections.viewCheckpoint(this.checkpointOf(id)) + } + /** * Durably checkpoint one live session NOW (both mandatory points call * this; tests and carriers may too). The registry cut is snapshotted at diff --git a/packages/session-projection/session-projection/src/index.ts b/packages/session-projection/session-projection/src/index.ts index c2974b565d..2f952166cc 100644 --- a/packages/session-projection/session-projection/src/index.ts +++ b/packages/session-projection/session-projection/src/index.ts @@ -280,6 +280,27 @@ export class SessionProjectionRegistry extends Service { return floor === undefined ? undefined : Math.max(floor - 1, 0) } + /** + * View a checkpoint's rows without any log read: for every registered + * unit whose row's `stateVersion` matches, serve the schema-validated + * `view` of the stored state; mismatched or absent rows leave their key + * absent (a cold or listing consumer treats it as not-yet-available and a + * fuller read path refolds it). The zero-I/O rung of the read ladder — + * values are as stale as their rows, never wrong. + * @param checkpoint - persisted rows for one session (possibly stale or empty). + * @returns whole values per key with a usable row; empty when none. + */ + viewCheckpoint(checkpoint: ProjectionCheckpoint): Partial { + const values: Record = {} + for (const registration of this.registrations.values()) { + const def = registration.def + const row = checkpoint[def.key] + if (row === undefined || row.stateVersion !== def.stateVersion) continue + values[def.key] = def.schema.parse(def.view(row.state)) + } + return values as Partial + } + /** * Cold read: fold every registered unit over a stored log suffix, seeding * each from its checkpoint row when usable — the one read recipe (cached diff --git a/packages/session-projection/session-projection/tests/registry.spec.ts b/packages/session-projection/session-projection/tests/registry.spec.ts index 3c069f91d3..fc167a1625 100644 --- a/packages/session-projection/session-projection/tests/registry.spec.ts +++ b/packages/session-projection/session-projection/tests/registry.spec.ts @@ -267,6 +267,19 @@ describe('SessionProjectionRegistry drive', () => { expect(current.values['test/count']).toBe(5) }) + it('viewCheckpoint serves version-matching rows without any log and skips mismatched keys', async () => { + const { ctx } = await harness() + ctx.sessionProjections.register(marksUnit()) + ctx.sessionProjections.register(countUnit()) + const values = ctx.sessionProjections.viewCheckpoint({ + 'test/marks': { stateVersion: 1, observedSeq: 4, state: { marks: ['stored'] } }, + 'test/count': { stateVersion: 99, observedSeq: 4, state: 5 }, // mismatched: absent + }) + expect(values['test/marks']).toEqual({ marks: ['stored'] }) + expect('test/count' in values).toBe(false) + expect(ctx.sessionProjections.viewCheckpoint({})).toEqual({}) + }) + it('restore rejects a row claiming events past the supplied log end (shrunk log ⇒ re-read)', async () => { const { ctx } = await harness() ctx.sessionProjections.register(countUnit()) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7e189366cb..408a990c35 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2685,6 +2685,9 @@ importers: '@deepseek-ai/dsh-session-projection': specifier: workspace:^ version: link:../../session-projection/session-projection + '@deepseek-ai/dsh-session-projection-cache': + specifier: workspace:^ + version: link:../../session-projection/session-projection-cache '@deepseek-ai/dsh-skill': specifier: workspace:^ version: link:../../skill/skill