diff --git a/packages/client/runtime/src/client/sessions/manager.ts b/packages/client/runtime/src/client/sessions/manager.ts index 66a0d00dd9..1e88e9f8ba 100644 --- a/packages/client/runtime/src/client/sessions/manager.ts +++ b/packages/client/runtime/src/client/sessions/manager.ts @@ -212,6 +212,19 @@ export class SessionManager { session.handleBlank(s.blank) session.handleRunning(s.running) } + // Seed each row's projection baseline into the per-session value + // store (cold titles surface without opening the session). Per-key + // apply, not seed(): the list block is a partial baseline — the + // cold cache serves only version-matching keys — so an absent key + // must not clear; higher-seq-wins still keeps a stale list block + // from overwriting a newer push frame or tail baseline. + for (const s of result.value.items) { + const block = s.projections + if (block === undefined) continue + const store = this.projectionStore(s.sessionId) + const values = block.values as Record + for (const key of Object.keys(values)) store.apply(key, values[key], block.asOfSeq) + } } else { this.listState = 'error' this.listError = result.error diff --git a/packages/client/runtime/tests/manager.spec.ts b/packages/client/runtime/tests/manager.spec.ts index 2923cd0d3c..85b9ed6b75 100644 --- a/packages/client/runtime/tests/manager.spec.ts +++ b/packages/client/runtime/tests/manager.spec.ts @@ -160,6 +160,28 @@ describe('list lifecycle', () => { expect(manager.getListSnapshot().items.find(item => item.sessionId === S1)?.title).toBeUndefined() }) + it('seeds cold titles from the list rows\' projections block under higher-seq-wins', async () => { + const api = new FakeApiClient() + const manager = new SessionManager(api) + // A push frame landed before the list (S2's title is newer than the block's cut). + manager.handleMuxEnvelope({ + rpcId: 'push-newer' as never, + payload: { type: 'session/projection', sessionId: S2, key: 'title', value: 'Pushed', seq: 9 } as never, + }) + api.onList = () => Promise.resolve(ok({ + items: [ + { ...summary(S1), projections: { asOfSeq: 4, values: { title: 'Cold cached' } } }, + { ...summary(S2, { updatedAt: 200 }), projections: { asOfSeq: 5, values: { title: 'List stale' } } }, + ] as never[], + })) + await manager.refreshList() + const items = manager.getListSnapshot().items + // Cold row: title surfaces straight from the list block — no open, no history. + expect(items.find(item => item.sessionId === S1)?.title).toBe('Cold cached') + // The stale list block (seq 5) cannot overwrite the newer push frame (seq 9). + expect(items.find(item => item.sessionId === S2)?.title).toBe('Pushed') + }) + it('drops a projection row beyond the subscription baseline before accepting its durable replay', async () => { const api = new FakeApiClient() api.onList = () => Promise.resolve(ok({ items: [summary(S1)] as never[] })) diff --git a/packages/host/apiproxy/src/api-proxy.ts b/packages/host/apiproxy/src/api-proxy.ts index 486a528a78..8526052a8b 100644 --- a/packages/host/apiproxy/src/api-proxy.ts +++ b/packages/host/apiproxy/src/api-proxy.ts @@ -29,7 +29,7 @@ import type { WorkspaceId, WorkspaceView, } from './api/index.ts' // Type-only: resolves `ctx.get('sessionProjections')` to the projection registry. -import type { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection' +import type {} 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')`. @@ -300,21 +300,23 @@ function projectionsFor(ctx: Context, agent: Agent): SessionProjectionsBlock | u } /** - * The projection column of one session.list row, fail-soft: attached + * The projection baseline 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. + * persisted projection cache's identity-checked stored rows (zero log loads + * either way — the listing use case the cache exists for). The block shape + * (values + asOfSeq) matches the history tail's, so a client seeds its + * value store under the same higher-seq-wins rule. Any failure — and an + * empty value set — yields an absent block: a listing without projections + * is degraded, never broken. */ -function listProjectionsFor(ctx: Context, id: SessionId, session: Session | undefined): Partial | undefined { +function listProjectionsFor(ctx: Context, meta: SessionHeader, session: Session | undefined): SessionProjectionsBlock | 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 + const block = session !== undefined + ? ctx.get('sessionProjections')?.snapshot(session) + : ctx.get('sessionProjectionCache')?.cachedSnapshot(meta) + return block !== undefined && Object.keys(block.values).length > 0 ? block : undefined } catch (error) { - ctx.logger.warn(`session.list: projection column for "${id}" failed (serving the row without it): ${String(error)}`) + ctx.logger.warn(`session.list: projection column for "${meta.id}" failed (serving the row without it): ${String(error)}`) return undefined } } @@ -676,7 +678,7 @@ 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) - const projections = listProjectionsFor(ctx, session.id, session) + const projections = listProjectionsFor(ctx, session.header, session) return { ...summarize(session, agent?.status === 'running'), ...projections === undefined ? {} : { projections }, @@ -689,7 +691,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro 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) + const projections = listProjectionsFor(ctx, meta, undefined) return { ...await summarizeCold(persistence, meta), ...projections === undefined ? {} : { projections }, diff --git a/packages/host/apiproxy/src/api/sessions.schema.ts b/packages/host/apiproxy/src/api/sessions.schema.ts index a02267a6bf..21feaaf604 100644 --- a/packages/host/apiproxy/src/api/sessions.schema.ts +++ b/packages/host/apiproxy/src/api/sessions.schema.ts @@ -7,7 +7,6 @@ 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 { @@ -38,16 +37,7 @@ 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. */ +/** SessionSummary row of session.list (`projections` reuses the history block's shape and schema). */ export const sessionSummarySchema = z.object({ sessionId: sessionIdSchema, updatedAt: z.number(), @@ -55,8 +45,8 @@ export const sessionSummarySchema = z.object({ blank: z.boolean(), parentSessionId: sessionIdSchema.optional(), cwd: z.string().optional(), - projections: projectionValuesSchema.optional(), -}) satisfies z.ZodType> + projections: z.lazy(() => sessionProjectionsBlockSchema).optional(), +}) as unknown as z.ZodType> /** session.list request payload (cursor is a reserved seat, unimplemented in v1). */ export const sessionListRequestSchema = z.object({ @@ -64,9 +54,9 @@ export const sessionListRequestSchema = z.object({ }) satisfies z.ZodType>> /** session.list response value. */ -export const sessionListValueSchema = z.object({ +export const sessionListValueSchema: z.ZodType>> = z.object({ items: z.array(sessionSummarySchema), -}) satisfies z.ZodType>> +}) /** session.create request payload (at most one of workspaceId / cwd). */ export const sessionCreateRequestSchema = z.object({ diff --git a/packages/host/apiproxy/src/api/sessions.ts b/packages/host/apiproxy/src/api/sessions.ts index a6d5c1517c..ea5d81171d 100644 --- a/packages/host/apiproxy/src/api/sessions.ts +++ b/packages/host/apiproxy/src/api/sessions.ts @@ -144,16 +144,17 @@ export interface SessionSummary { /** 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. + * Projection baseline for this row, 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 (`asOfSeq` says exactly how stale), never wrong, and directly + * seedable into the client's per-session value store under its + * higher-seq-wins rule (a list baseline can never overwrite a newer push + * frame). 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 + projections?: SessionProjectionsBlock } /** 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 caa957d05d..bcfc067ff5 100644 --- a/packages/host/apiproxy/tests/api-proxy-projections.spec.ts +++ b/packages/host/apiproxy/tests/api-proxy-projections.spec.ts @@ -126,14 +126,15 @@ describe('session.history projections block', () => { }) describe('session.list projections column', () => { - it('serves attached rows from the live registry cut', async () => { + it('serves attached rows from the live registry cut, watermarked for client seeding', 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' }) + expect(row?.projections?.values['test/last-user']).toEqual({ text: 'm0' }) + expect(row?.projections?.asOfSeq).toBe(session.seq - 1) }) it('omits the column entirely when no registry is mounted', async () => { @@ -158,13 +159,17 @@ describe('session.list projections column', () => { readFrom: load, } as never) ctx.provide('sessionProjectionCache', { - cachedValues: (id: unknown) => (id === coldId ? { 'test/last-user': { text: 'cached' } } : {}), + // The carrier hands the listed header through as the identity witness. + cachedSnapshot: (meta: { id: unknown; createdAt: number }) => + (meta.id === coldId && meta.createdAt === 5 + ? { asOfSeq: 7, values: { 'test/last-user': { text: 'cached' } } } + : 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?.running).toBe(false) - expect(row?.projections?.['test/last-user']).toEqual({ text: 'cached' }) + expect(row?.projections).toEqual({ asOfSeq: 7, values: { 'test/last-user': { text: 'cached' } } }) }) it('cold rows without a cache plugin (or without a stored row) just lack the column', async () => {