feat(session-projection-cache): seed cold reads from the cache and write back

A detached history read still traverses the complete log, but each unit's
fold is now seeded from its cached checkpoint: the registry's restore
slices off the already-folded prefix (events at or below the row's seq)
and applies only the tail. The first cold read writes the refreshed
checkpoint back (fail-soft), so the cache row is created on first read
and kept current afterwards. The recipe lives on the cache
(cachedCheckpoint, coldSnapshot, writeBack); the api-proxy carrier only
supplies the stored header and the full log.

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

@@ -821,14 +821,21 @@ function listProjectionsFor(
   }
 }
 
-/** Projection baseline for a detached history tail without Agent activation. */
+/**
+ * Projection baseline for a detached history tail without Agent activation.
+ * The cache owns the cold-read recipe (seed from the cached checkpoint,
+ * skip the cached prefix's applies, write the refreshed checkpoint back);
+ * without the cache the registry restores from init over the full log.
+ */
 function detachedProjectionsFor(
   ctx: Context,
+  meta: SessionHeader,
   events: readonly SessionEvent[],
 ): SessionProjectionsBlock | undefined {
   const registry = ctx.get('sessionProjections')
   if (registry === undefined) return undefined
-  return registry.restore({}, events, 0).snapshot
+  const cache = ctx.get('sessionProjectionCache')
+  return cache?.coldSnapshot(meta, events) ?? registry.restore({}, events, 0).snapshot
 }
 
 /**
@@ -1520,7 +1527,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
     includeProjections: boolean,
   ): { events: SessionEvent[]; projections?: SessionProjectionsBlock } {
     if (source.kind === 'detached') {
-      const projections = includeProjections ? detachedProjectionsFor(ctx, source.events) : undefined
+      const projections = includeProjections ? detachedProjectionsFor(ctx, source.header, source.events) : undefined
       return { events: source.events, ...projections === undefined ? {} : { projections } }
     }
     const events = [...source.session.events]
@@ -2625,7 +2632,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
             header = inspected.meta
             events = inspected.events
             projections = beforeSeq === undefined
-              ? subagentHistoryProjections(ctx, childSessionId, () => detachedProjectionsFor(ctx, inspected.events))
+              ? subagentHistoryProjections(ctx, childSessionId, () => detachedProjectionsFor(ctx, inspected.meta, inspected.events))
               : undefined
           } catch (error: unknown) {
             if (signal?.aborted) {

packages/host/apiproxy/tests/api-proxy-cold.spec.ts

@@ -15,6 +15,7 @@ import { TypertLookupFailure } from '@deepseek-ai/dsh-typert-protocol'
 import TypertRegistry from '@deepseek-ai/dsh-typert-registry'
 import { createUserMessage, MessageId } from '@deepseek-ai/dsh-llm'
 import type { Agent } from '@deepseek-ai/dsh-agent'
+import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
 import UserQuestionService from '@deepseek-ai/dsh-user-questions'
 import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
 import {
@@ -103,6 +104,9 @@ describe('sessions.list cold merge', () => {
         }
         return undefined
       },
+      // The detached cold-history path calls the cache's coldSnapshot; this
+      // listing harness never detaches, so it is inert here.
+      coldSnapshot: () => undefined,
     } as never)
     const api = createApiProxy(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' })
 
@@ -540,6 +544,45 @@ describe('subagent ownership fence', () => {
     expect(inspect).toHaveBeenCalledTimes(3)
   })
 
+  it('hands the detached cold history to the cache, which owns the seeded fold and write-back', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    await ctx.plugin(AgentRegistry)
+    await ctx.plugin(UserQuestionService)
+    await ctx.plugin(SessionProjectionRegistry)
+    const sessionId = sid('seeded-cold')
+    const meta = header(sessionId, 1)
+    const events = [
+      { type: 'user/message', seq: 0, time: 1, data: { content: [{ type: 'text', text: 'm0' }], source: { kind: 'user' } } },
+      { type: 'user/message', seq: 1, time: 2, data: { content: [{ type: 'text', text: 'm1' }], source: { kind: 'user' } } },
+      { type: 'user/message', seq: 2, time: 3, data: { content: [{ type: 'text', text: 'm2' }], source: { kind: 'user' } } },
+    ] as SessionEvent[]
+    const inspect = vi.fn(() => Promise.resolve({ meta, events }))
+    ctx.provide('sessionPersistence', {
+      list: () => Promise.resolve([meta]),
+      inspect,
+      locate: () => undefined,
+    } as never)
+    // The cache owns the cold-read recipe (seed, prefix-skip, write-back);
+    // the carrier's only job is to hand it the stored header and the full log.
+    const coldSnapshot = vi.fn((_meta: SessionHeader, _events: readonly SessionEvent[]) =>
+      ({ asOfSeq: 2, values: { 'test/last-user': { text: 'm2' } } }))
+    ctx.provide('sessionProjectionCache', {
+      cachedSnapshot: () => undefined,
+      coldSnapshot,
+    } as never)
+    const api = createApiProxy(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' })
+
+    const history = await api.sessions.history(request({ sessionId }))
+    expect(history.result.ok).toBe(true)
+    if (!history.result.ok) throw new Error('unreachable')
+    expect(history.result.value.projections?.values['test/last-user']).toEqual({ text: 'm2' })
+    // The full log (all events) was handed to the cache's cold snapshot.
+    expect(coldSnapshot).toHaveBeenCalledTimes(1)
+    expect(coldSnapshot.mock.calls[0]?.[0]?.id).toBe(sessionId)
+    expect(coldSnapshot.mock.calls[0]?.[1]?.map(event => event.seq)).toEqual([0, 1, 2])
+  })
+
   it('no longer treats a descriptor-only cold child without origin as subagent-owned', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)

packages/session/session-projection-cache/src/index.ts

@@ -131,6 +131,28 @@ export class SessionProjectionCache extends Service {
     return { asOfSeq, values }
   }
 
+  /**
+   * Cold-read one session's projections from its complete log. Each unit is
+   * seeded from the identity-checked cached rows — the registry skips `apply`
+   * for the already-folded prefix (events at or below the row's `seq`) — and
+   * the refreshed checkpoint is written back (fail-soft, fire-and-forget), so
+   * the first cold read creates the cache row and later ones seed from it.
+   * The caller supplies the complete log in seq order: this service never
+   * consults the persistence layer.
+   * @param meta - the stored session header (identity witness).
+   * @param events - the session's complete log, in seq order.
+   * @returns the projection cut at the log end.
+   */
+  coldSnapshot(meta: SessionHeader, events: readonly SessionEvent[]): ProjectionSnapshot {
+    const restored = this.ctx.sessionProjections.restore(this.recordFor(meta.id, identityOf(meta))?.rows ?? {}, events, 0)
+    // Refresh the row so the next cold read seeds from it; fail-soft and
+    // fire-and-forget — a failed write-back only costs a longer tail replay.
+    void this.put(meta.id, identityOf(meta), restored.checkpoint).catch((error: unknown) => {
+      this.ctx.logger.warn(`session projection cache: cold-read write-back for "${meta.id}" failed (cache stays stale): ${String(error)}`)
+    })
+    return restored.snapshot
+  }
+
   /**
    * Durably checkpoint one live session NOW (both mandatory points call
    * this; tests and carriers may too). The registry cut is snapshotted at

packages/session/session-projection-cache/tests/cache.spec.ts

@@ -35,6 +35,7 @@ declare module '@deepseek-ai/dsh-session-projection/types' {
   interface SessionProjectionStateMap {
     'cache-test/marks': MarksState
     'cache-test/marks2': Map<string, string>
+    'cache-test/count': number
   }
   interface SessionProjectionMap {
     'cache-test/marks': { marks: string[] }
@@ -311,3 +312,66 @@ describe('SessionProjectionCache listing read', () => {
     expect(cache.cachedSnapshot(headerOf(SessionId('malformed')))).toBeUndefined()
   })
 })
+
+describe('SessionProjectionCache cold-read seeding', () => {
+  it('coldSnapshot traverses the full log but applies only the events after each cached watermark', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
+    roots.push(root)
+    // A cached row covering the prefix through seq 2 (three applies folded).
+    await seedRecord(root, 'cold-snap', {
+      'cache-test/count': { ver: 1, seq: 2, val: 3 },
+    }, { createdAt: 9 })
+    const { cache, ctx } = await harness({ root })
+    const apply = vi.fn((_state: number, _event: SessionEvent) => 1)
+    ctx.sessionProjections.register({
+      key: 'cache-test/count',
+      stateSchema: z.number().int().nonnegative(),
+      init: () => 0,
+      apply,
+      stateVersion: 1,
+    } satisfies ProjectionDefinition<'cache-test/count', number>)
+    const meta = headerOf(SessionId('cold-snap'), 9)
+    const events = Array.from({ length: 5 }, (_, seq) => ({
+      type: 'cache-test/mark', seq, time: seq, data: { marks: [`m${seq}`] },
+    })) as SessionEvent[]
+    const snapshot = cache.coldSnapshot(meta, events)
+    // The full log was traversed, but the fold applied only seqs 3 and 4.
+    expect(apply).toHaveBeenCalledTimes(2)
+    expect(apply.mock.calls.map(call => call[1].seq)).toEqual([3, 4])
+    expect(snapshot.asOfSeq).toBe(4)
+    // Host-only unit: folded but not served; the refreshed row is written
+    // back (fail-soft, fire-and-forget) once the write lands.
+    expect(Object.keys(snapshot.values)).not.toContain('cache-test/count')
+    await settle()
+    expect((await storedRows(root, meta.id))?.['cache-test/count']?.seq).toBe(4)
+    // No cached row yet: the first cold read folds from init over the full
+    // log and creates the cache row (the `?? {}` seed path).
+    const fresh = headerOf(SessionId('cold-fresh'), 10)
+    cache.coldSnapshot(fresh, events)
+    expect(apply).toHaveBeenCalledTimes(7) // 2 tail + 5 full
+    await settle()
+    expect((await storedRows(root, fresh.id))?.['cache-test/count']?.seq).toBe(4)
+  })
+
+  it('coldSnapshot write-back is fail-soft: a failed durable write logs and never throws', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
+    roots.push(root)
+    const ctx = new Context()
+    contexts.push(ctx)
+    await ctx.plugin(Storage)
+    await ctx.plugin({ name: storageJsonName, inject: storageJsonInject, apply: storageJsonApply, Config: storageJsonConfig }, { root })
+    await ctx.plugin({ name: storageDomainName, inject: storageDomainInject, apply: storageDomainApply, Config: storageDomainConfig }, { backend: 'json' })
+    await ctx.plugin(SessionStore)
+    await ctx.plugin(SessionProjectionRegistry)
+    ctx.sessionProjections.register(marksUnit())
+    await ctx.plugin(SessionProjectionCache, { writeEveryEvents: 100, writeIntervalMs: 60_000 })
+    const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
+    // A directory where the record document must land makes the write-back
+    // fail; the cold read itself still succeeds and never throws.
+    const meta = headerOf(SessionId('cold-fail'))
+    await mkdir(recordPath(root, meta.id), { recursive: true })
+    expect(ctx.sessionProjectionCache.coldSnapshot(meta, [])).toBeDefined()
+    await settle()
+    expect(warn).toHaveBeenCalledWith(expect.stringContaining('cold-read write-back for "cold-fail" failed'))
+  })
+})

packages/session/session-projection/src/index.ts

@@ -440,10 +440,12 @@ export class SessionProjectionRegistry extends Service {
         )
       }
       let state = usable ? def.stateSchema.parse(row.val) : def.init()
+      // The events are seq-ordered: slice off the already-folded prefix
+      // (events at or below the seed watermark) and fold only the tail.
       const from = usable ? row.seq : baseSeq - 1
-      for (const event of events) {
-        if (event.seq > from) state = def.apply(state, event)
-      }
+      const tailStart = events.findIndex(event => event.seq > from)
+      const tail = tailStart === -1 ? [] : events.slice(tailStart)
+      for (const event of tail) state = def.apply(state, event)
       if (def.wire !== undefined) values[def.key] = def.wire.viewSchema.parse(def.wire.view(state))
       refreshed[def.key] = { ver: def.stateVersion, seq: endSeq, val: state }
     }