fix(host): fold monster sessions in slices during projection backfill (issue #145)

A whole-log cold restore lands on the host event loop as one uninterruptible
synchronous block (~1s per 30k events on a fast host, more on a Windows
desktop), and a 100+ session backfill strings those into a window the desktop
renderer watchdog cannot survive (issue #145: 118 sessions, 47s, renderer
killed).

Sessions over 2,000 events now fold in 2,000-event in-memory slices through
the public registry.restore, yielding between slices, with exactly one durable
write-back at the end (sessions.prepare + registry.hydrate with the rows
already at the log end, then sessionProjectionCache.write). The composed rows
are byte-identical to one single-shot coldSnapshot (asserted with the real
registry + the plugin's real units), no intermediate durable rows land, and a
crash mid-fold keeps the single-shot failure semantics. A session going live
mid-fold skips silently; a write-back failure stays fail-soft.

Faces are re-proved per deployment and any absence (older/unknown harness)
keeps the single-shot coldSnapshot path; all four faces exist on both
supported baselines (0.1.7-rc.2, 0.2.0-rc.2), verified against the tags.

src/host/backfill.ts

@@ -40,6 +40,18 @@ interface ProjectionCacheLike {
   coldSnapshot(header: SessionHeader, inheritedEventCount: unknown, events: readonly SessionEvent[]): unknown
 }
 
+/** The sliced fold's extra faces (all re-proved at runtime; any absence keeps the
+ * single-shot path). `restore` and `hydrate` ride the registry the plugin already injects;
+ * `prepare` and `write` are the same faces session-query's observation handoff uses. */
+interface SlicedFoldFaces {
+  restore(
+    checkpoint: unknown, events: readonly SessionEvent[], baseSeq: number, header: SessionHeader, inheritedEventCount: unknown,
+  ): unknown
+  hydrate(session: unknown, checkpoint: unknown, events: readonly SessionEvent[], baseSeq: number): unknown
+  prepare(id: string, options: { seed: readonly SessionEvent[]; meta: unknown; inheritedEventCount: unknown; eventState: string }): unknown
+  write(session: unknown): Promise<unknown>
+}
+
 /** One persistence read handle, as consumed. */
 interface ReadHandleLike {
   header: SessionHeader
@@ -56,6 +68,12 @@ interface PersistenceLike {
  * sustained read pressure stalls its own client handshake. */
 const YIELD_MS = 100
 
+/** The sliced fold's granularity and inter-slice pause: at ~13µs/event a slice is a ~25ms
+ * synchronous block (measured ≤88ms including the per-slice state re-validation), so even a
+ * 30k-event session never lands on the event loop as one uninterruptible second. */
+const SLICE_EVENTS = 2_000
+const SLICE_YIELD_MS = 25
+
 /** The `name` family dsh raises for a durable log it cannot interpret: the persistence seam
  * translates the format edge's refusal into `SessionFormatUnsupportedError`, so classify by prefix and import no dsh symbol. */
 const UNSUPPORTED_FORMAT = 'SessionFormatUnsupported'
@@ -123,6 +141,60 @@ function isLive(sessions: Record<string, unknown> | null, id: string): boolean {
   }
 }
 
+/** The sliced fold's per-session outcome: the rows landed, the session went live mid-fold
+ * (it folds for itself — skip silently), or the pass was aborted mid-fold (stop quietly). */
+type SliceOutcome = 'done' | 'live' | 'aborted'
+
+/** One cold fold in event-count slices (issue #145): a monster session's whole-log restore
+ * would otherwise land as ONE uninterruptible synchronous block (measured ~1s per 30k events
+ * on a fast host) inside a boot-time window already crowded with a hundred other folds.
+ * Each slice seeds every unit from the previous slice's rows IN MEMORY (the registry's
+ * restore is the cache's own cold-read computation, so the composed result is byte-identical
+ * to one single-shot coldSnapshot), the event loop breathes between slices, and exactly ONE
+ * durable write-back lands at the end — a crash mid-fold leaves no partial rows, the same
+ * failure semantics as the single-shot path. */
+async function coldSnapshotSliced(
+  ctx: Context,
+  faces: SlicedFoldFaces,
+  sessions: Record<string, unknown> | null,
+  log: { header: SessionHeader; inheritedEventCount: unknown; events: readonly SessionEvent[] },
+  signal: AbortSignal,
+): Promise<SliceOutcome> {
+  const { header, inheritedEventCount, events } = log
+  let rows: unknown = {}
+  for (let end = SLICE_EVENTS; ; end += SLICE_EVENTS) {
+    const last = end >= events.length
+    const restored = asRecord(faces.restore(rows, last ? events : events.slice(0, end), 0, header, inheritedEventCount))
+    rows = restored?.checkpoint ?? {}
+    if (last) break
+    await new Promise(resolve => setTimeout(resolve, SLICE_YIELD_MS))
+    if (signal.aborted) return 'aborted'
+  }
+  let prepared: unknown
+  try {
+    prepared = faces.prepare(header.id, {
+      seed: events,
+      meta: structuredClone(header),
+      inheritedEventCount,
+      eventState: 'shared-frozen',
+    })
+  } catch (error: unknown) {
+    // The store rejects an id with a live owner: the session attached mid-fold and now folds for itself.
+    if (isLive(sessions, header.id)) return 'live'
+    throw error
+  }
+  // The rows end at the log's last seq, so hydrate installs the cells without refolding.
+  faces.hydrate(prepared, rows, events, 0)
+  try {
+    await faces.write(prepared)
+  } catch (error: unknown) {
+    // Fail-soft, mirroring the cache's own write-back posture: a lost write costs a refold
+    // on the next pass, never a wrong row.
+    ctx.logger.warn(`dsh-context: projection backfill write-back for "${header.id}" failed (cache stays stale): ${messageOf(error)}`)
+  }
+  return 'done'
+}
+
 function isUnsupportedFormat(error: unknown): boolean {
   try {
     return String(asRecord(error)?.name).startsWith(UNSUPPORTED_FORMAT)
@@ -179,6 +251,24 @@ export function watchActivityBackfill(ctx: Context, coldReads: ColdReadGate = ma
     if (persistence === null || typeof persistence.open !== 'function') return
     const sessions = asRecord(injected.sessions)
 
+    // The sliced fold's faces: the registry the plugin already injects, plus the sessions and
+    // cache write verbs. Any absence keeps every session on the single-shot coldSnapshot path.
+    // Bind at extraction: an unbound hand-off loses `this` on the real faces.
+    const registry = asRecord((ctx as unknown as Record<string, unknown>).sessionProjections)
+    const sliceFaces: SlicedFoldFaces | null = registry !== null
+      && typeof registry.restore === 'function'
+      && typeof registry.hydrate === 'function'
+      && sessions !== null
+      && typeof sessions.prepare === 'function'
+      && typeof cache.write === 'function'
+      ? {
+        restore: (registry.restore as SlicedFoldFaces['restore']).bind(registry),
+        hydrate: (registry.hydrate as SlicedFoldFaces['hydrate']).bind(registry),
+        prepare: (sessions.prepare as SlicedFoldFaces['prepare']).bind(sessions),
+        write: (cache.write as SlicedFoldFaces['write']).bind(cache),
+      }
+      : null
+
     const abort = new AbortController()
     const run = async (): Promise<void> => {
       const listed = await (query as unknown as SessionQueryLike).listSessions(abort.signal)
@@ -196,15 +286,19 @@ export function watchActivityBackfill(ctx: Context, coldReads: ColdReadGate = ma
           const admitted = await coldReads.admit(async () => {
             const log = await readColdLog(persistence as unknown as PersistenceLike, header.id, abort.signal)
             // The handle's header is authoritative (fixed at open); the listed one was its probe.
+            if (sliceFaces !== null && log.events.length > SLICE_EVENTS) {
+              return coldSnapshotSliced(ctx, sliceFaces, sessions, log, abort.signal)
+            }
             ;(cache as unknown as ProjectionCacheLike).coldSnapshot(log.header, log.inheritedEventCount, log.events)
-            return true
+            return 'done' as const
           })
           if (admitted === undefined) {
             // Pressure does not fall within a pacing tick, so re-probing per session only delays the host's own work.
             paused = true
             break
           }
-          folded++
+          if (admitted === 'aborted') return
+          if (admitted === 'done') folded++
         } catch (error: unknown) {
           if (isUnsupportedFormat(error)) {
             // Permanent and source-side: the per-session detail drops to

tests/host/backfill.spec.ts

@@ -3,8 +3,12 @@
 import assert from 'node:assert/strict'
 import { describe, test } from 'vitest'
 import { Context } from '@deepseek-ai/cordis'
+import { SessionProjectionRegistry } from '@deepseek-ai/dsh-session-projection'
 import type { SessionEvent } from '@deepseek-ai/dsh-session'
 import { BACKFILL_ROUTE, watchActivityBackfill } from '../../src/host/backfill'
+import { createContextTimelineDefinition } from '../../src/host/timeline'
+import { createContextActivityDefinition } from '../../src/host/activity'
+import { createContextHeadersDefinition } from '../../src/host/headers'
 
 async function until<T>(read: () => T | undefined, message: string): Promise<T> {
   for (let i = 0; i < 300; i++) {
@@ -761,3 +765,212 @@ describe('watchActivityBackfill', () => {
     assert.equal(lines.filter(l => l.text.includes('heap pressure')).length, 1, 'one line, not one per session')
   })
 })
+
+/** A contiguous fake log of the given length (restore validates seq continuity, nothing else). */
+function fakeEvents(length: number): SessionEvent[] {
+  return Array.from({ length }, (_, seq) => ({ type: 'todo/write', seq, time: seq + 1, data: {} })) as unknown as SessionEvent[]
+}
+
+interface SliceRig {
+  route: RouteBox
+  writes: Map<string, unknown>
+  prepares: string[]
+  coldSnapshots: string[]
+  restoreCalls: number[]
+}
+
+/** One single-shot restore over the same log on a fresh registry — the sliced path's reference answer. */
+function referenceRows(events: SessionEvent[], id: string): unknown {
+  const ctx = new Context()
+  const registry = new SessionProjectionRegistry(ctx as never)
+  registry.register(createContextTimelineDefinition({}, () => true) as never)
+  registry.register(createContextActivityDefinition() as never)
+  registry.register(createContextHeadersDefinition() as never)
+  return registry.restore({}, events, 0 as never, { id, version: 1, createdAt: 1, cwd: `/repo/${id}`, isSeeded: false } as never, 0 as never).checkpoint
+}
+
+/** Arm the backfill with the REAL registry + the plugin's REAL units, fake persistence serving
+ * `events`, and the sliced path's faces (prepare/write) wired to the registry the way dsh's
+ * SessionStore and SessionProjectionCache wire them. */
+function armSliced(ctx: Context, eventsById: Map<string, SessionEvent[]>, opts: {
+  liveOnPrepare?: Set<string>
+  failPrepare?: Set<string>
+  failWrite?: boolean
+  noPrepare?: boolean
+  hostileRestoreAt?: number
+} = {}): SliceRig {
+  const route: RouteBox = {}
+  const state = fakeState([...eventsById.keys()].map(id => ({ header: { id, cwd: `/repo/${id}` } })))
+  const rig: SliceRig = {
+    route,
+    writes: new Map(),
+    prepares: [],
+    coldSnapshots: state.coldSnapshots,
+    restoreCalls: [],
+  }
+  const registry = new SessionProjectionRegistry(ctx as never)
+  registry.register(createContextTimelineDefinition({}, () => true) as never)
+  registry.register(createContextActivityDefinition() as never)
+  registry.register(createContextHeadersDefinition() as never)
+  const realRestore = registry.restore.bind(registry)
+  registry.restore = ((...args: Parameters<typeof realRestore>) => {
+    rig.restoreCalls.push((args[1] as SessionEvent[]).length)
+    if (opts.hostileRestoreAt !== undefined && rig.restoreCalls.length === opts.hostileRestoreAt) return undefined
+    return realRestore(...args)
+  }) as typeof registry.restore
+  ctx.provide('connection', connectionOf(route))
+  ctx.provide('sessionQuery', { listSessions: async () => state.listed })
+  ctx.provide('sessionProjectionCache', {
+    cachedSnapshot: () => undefined,
+    coldSnapshot: (header: { id: string }) => {
+      state.coldSnapshots.push(header.id)
+      return { asOfSeq: 0, values: {} }
+    },
+    write: async (session: { id: string }) => {
+      if (opts.failWrite === true) throw new Error('disk full')
+      rig.writes.set(session.id, registry.checkpoint(session as never))
+    },
+  })
+  ctx.provide('sessionPersistence', {
+    open: async (id: string) => {
+      state.coldReads.push(id)
+      return {
+        header: { id, version: 1, createdAt: 1, cwd: `/repo/${id}`, isSeeded: false },
+        inheritedEventCount: 0,
+        read: async () => ({ events: eventsById.get(id) ?? [] }),
+        close: async () => {},
+      }
+    },
+  })
+  const threwPrepare = new Set<string>()
+  const sessionsFace: Record<string, unknown> = {
+    get: (id: string) => (state.live.has(id) || threwPrepare.has(id) ? { id } : undefined),
+  }
+  if (opts.noPrepare !== true) {
+    sessionsFace.prepare = (id: string, o: { seed: SessionEvent[]; meta: unknown; inheritedEventCount: number }) => {
+      rig.prepares.push(id)
+      if (opts.liveOnPrepare?.has(id) === true) {
+        threwPrepare.add(id)
+        throw new Error(`session "${id}" already exists`)
+      }
+      if (opts.failPrepare?.has(id) === true) throw new Error(`session "${id}" has a corrupt header`)
+      return {
+        id,
+        header: o.meta,
+        inheritedEventCount: o.inheritedEventCount,
+        seq: o.seed.length,
+        snapshotEvents: () => o.seed,
+      }
+    }
+  }
+  ctx.provide('sessions', sessionsFace)
+  return rig
+}
+
+describe('the sliced cold fold (issue #145)', () => {
+  test('a monster session folds in slices and writes back rows byte-identical to one single-shot restore', async () => {
+    const ctx = new Context()
+    const events = fakeEvents(5_001) // three slices: 2000 + 2000 + 1001
+    const rig = armSliced(ctx, new Map([['monster', events]]))
+    const dispose = watchActivityBackfill(ctx)
+    await trigger(rig.route)
+    await until(() => (rig.writes.has('monster') ? true : undefined), 'the write-back landed')
+    dispose()
+    assert.deepEqual(rig.coldSnapshots, [], 'the sliced path never calls coldSnapshot')
+    assert.deepEqual(rig.prepares, ['monster'])
+    assert.ok(rig.restoreCalls.length >= 3, `expected ≥3 sliced restores, got ${rig.restoreCalls.join(',')}`)
+    assert.deepStrictEqual(rig.writes.get('monster'), referenceRows(events, 'monster'), 'sliced rows must equal the single-shot rows byte for byte')
+  })
+
+  test('a hostile restore answer (no checkpoint) refolds from init and still lands the exact rows', async () => {
+    const ctx = new Context()
+    const events = fakeEvents(4_001)
+    const rig = armSliced(ctx, new Map([['monster', events]]), { hostileRestoreAt: 2 })
+    const dispose = watchActivityBackfill(ctx)
+    await trigger(rig.route)
+    await until(() => (rig.writes.has('monster') ? true : undefined), 'the write-back landed')
+    dispose()
+    assert.deepStrictEqual(rig.writes.get('monster'), referenceRows(events, 'monster'), 'a dropped slice answer degrades to a from-init refold — same rows')
+  })
+
+  test('a small session stays on the single-shot coldSnapshot path even when the slice faces exist', async () => {
+    const ctx = new Context()
+    const rig = armSliced(ctx, new Map([['small', fakeEvents(2_000)]]))
+    const dispose = watchActivityBackfill(ctx)
+    await trigger(rig.route)
+    await until(() => (rig.coldSnapshots.length === 1 ? true : undefined), 'the coldSnapshot ran')
+    dispose()
+    assert.deepEqual(rig.coldSnapshots, ['small'])
+    assert.deepEqual(rig.prepares, [], 'no prepare for a session at or under the slice threshold')
+    assert.deepEqual([...rig.writes.keys()], [])
+  })
+
+  test('missing slice faces fall back to coldSnapshot for the same monster session', async () => {
+    const ctx = new Context()
+    const rig = armSliced(ctx, new Map([['monster', fakeEvents(4_001)]], ), { noPrepare: true })
+    const dispose = watchActivityBackfill(ctx)
+    await trigger(rig.route)
+    await until(() => (rig.coldSnapshots.length === 1 ? true : undefined), 'the fallback ran')
+    dispose()
+    assert.deepEqual(rig.coldSnapshots, ['monster'], 'no prepare face → single-shot path')
+    assert.deepEqual([...rig.writes.keys()], [])
+  })
+
+  test('a session going live mid-fold is skipped silently (no write, no warn)', async () => {
+    const ctx = new Context()
+    const rig = armSliced(ctx, new Map([['racer', fakeEvents(4_001)]]), { liveOnPrepare: new Set(['racer']) })
+    const lines = captureLogs(ctx)
+    const dispose = watchActivityBackfill(ctx)
+    await trigger(rig.route)
+    await new Promise(resolve => setTimeout(resolve, 250))
+    dispose()
+    assert.deepEqual(rig.prepares, ['racer'])
+    assert.deepEqual([...rig.writes.keys()], [], 'no write over a session that went live')
+    assert.deepEqual(lines.filter(l => l.level === 'warn'), [], 'a live race never warns')
+  })
+
+  test('a prepare failure with the session still cold warns and the run continues', async () => {
+    const ctx = new Context()
+    const rig = armSliced(ctx, new Map([['broken', fakeEvents(4_001)], ['good', fakeEvents(10)]]), { failPrepare: new Set(['broken']) })
+    const lines = captureLogs(ctx)
+    const dispose = watchActivityBackfill(ctx)
+    await trigger(rig.route)
+    await until(() => (rig.coldSnapshots.length === 1 ? true : undefined), 'the small session folded')
+    await new Promise(resolve => setTimeout(resolve, 250))
+    dispose()
+    assert.deepEqual([...rig.writes.keys()], [])
+    const warns = lines.filter(l => l.level === 'warn')
+    assert.equal(warns.length, 1, 'the genuine prepare failure keeps the per-session warn')
+    assert.match(warns[0].text, /"broken"/)
+    assert.match(warns[0].text, /corrupt header/)
+    assert.deepEqual(rig.coldSnapshots, ['good'], 'the next session still folds')
+  })
+
+  test('a rejecting write-back warns (fail-soft) and the session still counts as folded', async () => {
+    const ctx = new Context()
+    const rig = armSliced(ctx, new Map([['monster', fakeEvents(4_001)]]), { failWrite: true })
+    const lines = captureLogs(ctx)
+    const dispose = watchActivityBackfill(ctx)
+    await trigger(rig.route)
+    await until(() => (lines.some(l => l.text.includes('write-back')) ? true : undefined), 'the write-back warning')
+    await new Promise(resolve => setTimeout(resolve, 250))
+    dispose()
+    const warns = lines.filter(l => l.level === 'warn')
+    assert.equal(warns.length, 1)
+    assert.match(warns[0].text, /write-back for "monster" failed/)
+    assert.ok(lines.some(l => l.level === 'info' && l.text.includes('backfilled for 1 session(s)')), 'the fold itself succeeded')
+  })
+
+  test('dispose mid-fold aborts between slices: no write, no completion line', async () => {
+    const ctx = new Context()
+    const rig = armSliced(ctx, new Map([['monster', fakeEvents(6_001)]]) )
+    const lines = captureLogs(ctx)
+    const dispose = watchActivityBackfill(ctx)
+    await trigger(rig.route)
+    await until(() => (rig.restoreCalls.length >= 1 ? true : undefined), 'the first slice folded')
+    dispose()
+    await new Promise(resolve => setTimeout(resolve, 250))
+    assert.deepEqual([...rig.writes.keys()], [], 'nothing durable lands mid-fold')
+    assert.deepEqual(lines.filter(l => l.level === 'info' && l.text.includes('backfilled')), [], 'the pass stopped quietly')
+  })
+})