fix(apiproxy,client): the list projection column becomes a seedable watermarked block

Review finding (PR #791): the column carried bare values (no seq), so the
client could not seed its value store without risking a stale list block
outranking newer push frames — and nothing consumed the column at all,
leaving cold titles absent after a restart. SessionSummary.projections is
now the same SessionProjectionsBlock as the history tail (values +
asOfSeq; attached rows cut the live registry, cold rows serve the cache's
identity-checked cachedSnapshot whose asOfSeq is the lowest served-row
watermark). SessionManager.refreshList seeds each row's block into the
per-session projection store via per-key apply — partial-baseline
semantics: an absent key never clears, and higher-seq-wins keeps stale
list blocks beneath push frames and tail baselines — so cold titles
surface in the sidebar without opening a session.

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<string, unknown>
+            for (const key of Object.keys(values)) store.apply(key, values[key], block.asOfSeq)
+          }
         } else {
           this.listState = 'error'
           this.listError = result.error

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[] }))

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<SessionProjectionMap> | 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 },

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<SessionEvent>
 
-/**
- * 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<Partial<SessionProjectionMap>>
-
-/** 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<Wire<SessionSummary>>
+  projections: z.lazy(() => sessionProjectionsBlockSchema).optional(),
+}) as unknown as z.ZodType<Wire<SessionSummary>>
 
 /** 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<Wire<RequestPayload<'session.list'>>>
 
 /** session.list response value. */
-export const sessionListValueSchema = z.object({
+export const sessionListValueSchema: z.ZodType<Wire<ResponseValue<'session.list'>>> = z.object({
   items: z.array(sessionSummarySchema),
-}) satisfies z.ZodType<Wire<ResponseValue<'session.list'>>>
+})
 
 /** session.create request payload (at most one of workspaceId / cwd). */
 export const sessionCreateRequestSchema = z.object({

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<SessionProjectionMap>
+  projections?: SessionProjectionsBlock
 }
 
 /** Session-domain unary methods (the map keys session.* of RpcMethodMap). */

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 () => {