Merge origin/master (regenerate the event producer-consumer graph)

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:^",

packages/host/apiproxy/src/api-proxy.ts

@@ -30,6 +30,8 @@ import type {
 } from './api/index.ts'
 // Type-only: resolves `ctx.get('sessionProjections')` to the projection registry.
 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')`.
 import type {} from '@deepseek-ai/dsh-commands'
 import type {} from '@deepseek-ai/dsh-skill'
@@ -309,6 +311,28 @@ function projectionsFor(ctx: Context, agent: Agent): SessionProjectionsBlock | u
   return registry.snapshot(agent.session)
 }
 
+/**
+ * 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 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, meta: SessionHeader, session: Session | undefined): SessionProjectionsBlock | undefined {
+  try {
+    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 "${meta.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).
@@ -666,13 +690,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.header, 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, undefined)
+            return {
+              ...await summarizeCold(persistence, meta),
+              ...projections === undefined ? {} : { projections },
+            }
+          })))
         }
         items.sort((a, b) => b.updatedAt - a.updatedAt)
         return ok(request, { items })

packages/host/apiproxy/src/api/sessions.schema.ts

@@ -37,7 +37,7 @@ export const sessionEventSchema = z.object({
   surfaceOp: z.unknown().optional(),
 }) as unknown as z.ZodType<SessionEvent>
 
-/** 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(),
@@ -45,7 +45,8 @@ export const sessionSummarySchema = z.object({
   blank: z.boolean(),
   parentSessionId: sessionIdSchema.optional(),
   cwd: z.string().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({
@@ -53,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

@@ -145,6 +145,18 @@ export interface SessionSummary {
   parentSessionId?: SessionId
   /** Session working directory (header.cwd passthrough); absent when unrecorded. */
   cwd?: string
+  /**
+   * 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?: SessionProjectionsBlock
 }
 
 /** Session-domain unary methods (the map keys session.* of RpcMethodMap). */

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,82 @@ describe('session.history projections block', () => {
   })
 })
 
+describe('session.list projections column', () => {
+  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?.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 () => {
+    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', {
+      // 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).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 () => {
+    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<RpcRequest<MuxFrame>>, count: number, abort: AbortController): Promise<MuxFrame[]> {

packages/host/apiproxy/tsconfig.json

@@ -35,6 +35,9 @@
     {
       "path": "../../session-projection/session-projection"
     },
+    {
+      "path": "../../session-projection/session-projection-cache"
+    },
     {
       "path": "../../skill/skill"
     },