feat(host): pace the projection backfill proportionally with a logged work budget (issue #145)

The pass used to hold the host at ~60% duty for the whole corpus window (a
fixed 100ms sleep between full-log folds). Rests are now proportional to the
measured work (at least a 25ms floor) at both granularities — between slices
and between sessions — capping the pass near 50% duty, and every 8s of
cumulative fold work the pass takes a logged 2s recovery pause and resumes on
its own (the issue's 'bounded, background, visibly paused' ask). Slice work is
billed exactly once across the two granularities, and every sleep wakes early
on abort so dispose never waits out a pause.

Pacing is injectable (BackfillPacing) for deterministic tests, mirroring the
cold-read gate's injectable reading; the sliced-fold suite moves to its own
spec file with the shared rig extracted to tests/host/helpers/backfill.ts.
What gets folded and the final cache rows are unchanged — only when the loop
breathes.

src/host/backfill.ts

@@ -20,6 +20,7 @@
  */
 
 import { asRecord } from './record'
+import { performance } from 'node:perf_hooks'
 import type { Context } from '@deepseek-ai/cordis'
 import { interruptedTurnClosers } from '@deepseek-ai/dsh-session'
 import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
@@ -64,15 +65,39 @@ interface PersistenceLike {
   open(id: string, access: 'read', options?: { signal?: AbortSignal }): Promise<ReadHandleLike>
 }
 
-/** Inter-session pacing: each cold read is a full log decode, and a boot-time host under
- * sustained read pressure stalls its own client handshake. */
-const YIELD_MS = 100
+/** The pacing floor for BOTH granularities (between slices and between sessions): each cold
+ * read is a full log decode, and a boot-time host under sustained read pressure stalls its own
+ * client handshake. The actual rest is proportional to the work just done (at least this
+ * floor), so the pass never exceeds ~50% duty (issue #145). */
+const MIN_REST_MS = 25
 
-/** 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. */
+/** The sliced fold's granularity: 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 pass's sustained-work budget: after this much cumulative fold time the pass takes a
+ * full recovery pause (and says so), then resumes where it stopped — a long corpus fills
+ * progressively instead of holding the host in one crowded window. */
+const ROUND_WORK_BUDGET_MS = 8_000
+const ROUND_PAUSE_MS = 2_000
+
+/** The pacing seam, injectable for tests (mirroring the cold-read gate's `reading`). */
+export interface BackfillPacing {
+  now: () => number
+  sleep: (ms: number) => Promise<void>
+  minRestMs: number
+  workBudgetMs: number
+  roundPauseMs: number
+}
+
+const DEFAULT_PACING: BackfillPacing = {
+  now: () => performance.now(),
+  sleep: ms => new Promise(resolve => setTimeout(resolve, ms)),
+  minRestMs: MIN_REST_MS,
+  workBudgetMs: ROUND_WORK_BUDGET_MS,
+  roundPauseMs: ROUND_PAUSE_MS,
+}
 
 /** 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. */
@@ -142,8 +167,12 @@ 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'
+ * (it folds for itself — skip silently), or the pass was aborted mid-fold (stop quietly).
+ * `workMs` is the measured slice work, so the session-level accounting never bills it twice. */
+interface SliceOutcome {
+  outcome: 'done' | 'live' | 'aborted'
+  workMs: number
+}
 
 /** 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
@@ -159,16 +188,22 @@ async function coldSnapshotSliced(
   sessions: Record<string, unknown> | null,
   log: { header: SessionHeader; inheritedEventCount: unknown; events: readonly SessionEvent[] },
   signal: AbortSignal,
+  rest: (costMs: number) => Promise<void>,
+  now: () => number,
 ): Promise<SliceOutcome> {
   const { header, inheritedEventCount, events } = log
   let rows: unknown = {}
+  let workMs = 0
   for (let end = SLICE_EVENTS; ; end += SLICE_EVENTS) {
     const last = end >= events.length
+    const started = now()
     const restored = asRecord(faces.restore(rows, last ? events : events.slice(0, end), 0, header, inheritedEventCount))
     rows = restored?.checkpoint ?? {}
+    const sliceCost = now() - started
+    workMs += sliceCost
     if (last) break
-    await new Promise(resolve => setTimeout(resolve, SLICE_YIELD_MS))
-    if (signal.aborted) return 'aborted'
+    await rest(sliceCost)
+    if (signal.aborted) return { outcome: 'aborted', workMs }
   }
   let prepared: unknown
   try {
@@ -180,7 +215,7 @@ async function coldSnapshotSliced(
     })
   } 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'
+    if (isLive(sessions, header.id)) return { outcome: 'live', workMs }
     throw error
   }
   // The rows end at the log's last seq, so hydrate installs the cells without refolding.
@@ -192,7 +227,7 @@ async function coldSnapshotSliced(
     // 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'
+  return { outcome: 'done', workMs }
 }
 
 function isUnsupportedFormat(error: unknown): boolean {
@@ -215,7 +250,11 @@ function messageOf(error: unknown): string {
 /** Arm the warm-up behind its trigger route: the route flips `requested` on the page's
  * first open and the cold-path faces compose `launch`; whichever lands second launches the one
  * pass. Returns the deferred injects' disposer (abort on unload). */
-export function watchActivityBackfill(ctx: Context, coldReads: ColdReadGate = makeColdReadGate()): () => void {
+export function watchActivityBackfill(
+  ctx: Context,
+  coldReads: ColdReadGate = makeColdReadGate(),
+  pacing: BackfillPacing = DEFAULT_PACING,
+): () => void {
   let requested = false
   let started = false
   let launch: (() => void) | null = null
@@ -276,18 +315,53 @@ export function watchActivityBackfill(ctx: Context, coldReads: ColdReadGate = ma
       let folded = 0
       let refused = 0
       let paused = false
+      // One rest serves both granularities (between slices and between sessions): the pause is
+      // proportional to the work just done, and a full recovery break follows every workBudgetMs
+      // of cumulative fold time — the pass says when it pauses and resumes on its own. The sleep
+      // wakes early on abort so dispose never waits out a pause.
+      let roundWork = 0
+      const sleep = async (ms: number): Promise<void> => {
+        if (abort.signal.aborted) return
+        // Assigned synchronously by the Promise executor below.
+        let onAbort!: () => void
+        try {
+          await Promise.race([
+            pacing.sleep(ms),
+            new Promise<void>((resolve) => {
+              onAbort = () => { resolve() }
+              abort.signal.addEventListener('abort', onAbort, { once: true })
+            }),
+          ])
+        } finally {
+          abort.signal.removeEventListener('abort', onAbort)
+        }
+      }
+      const rest = async (costMs: number): Promise<void> => {
+        roundWork += costMs
+        if (roundWork >= pacing.workBudgetMs) {
+          roundWork = 0
+          ctx.logger.info(`dsh-context: projection backfill pausing after ${Math.round(pacing.workBudgetMs / 1000)}s of fold work; resuming in ${Math.round(pacing.roundPauseMs / 1000)}s`)
+          await sleep(pacing.roundPauseMs)
+          return
+        }
+        await sleep(Math.max(pacing.minRestMs, costMs))
+      }
       for (const record of listed) {
         if (abort.signal.aborted) return
         const header = headerOf(record)
         if (header === null || typeof header.cwd !== 'string') continue
         if (isLive(sessions, header.id)) continue
         if (servesRows(cache as unknown as ProjectionCacheLike, header)) continue
+        const started = pacing.now()
+        let slicedWorkMs = 0
         try {
           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)
+              const sliced = await coldSnapshotSliced(ctx, sliceFaces, sessions, log, abort.signal, rest, pacing.now)
+              slicedWorkMs = sliced.workMs
+              return sliced.outcome
             }
             ;(cache as unknown as ProjectionCacheLike).coldSnapshot(log.header, log.inheritedEventCount, log.events)
             return 'done' as const
@@ -309,7 +383,9 @@ export function watchActivityBackfill(ctx: Context, coldReads: ColdReadGate = ma
             ctx.logger.warn(`dsh-context: projection backfill skipped "${header.id}" (${messageOf(error)})`)
           }
         }
-        await new Promise(resolve => setTimeout(resolve, YIELD_MS))
+        // Slice work already rested (and was billed) mid-fold; the session-level rest covers the
+        // remainder only — decode, write-back, or the whole single-shot fold.
+        await rest(Math.max(0, pacing.now() - started - slicedWorkMs))
       }
       if (folded > 0) ctx.logger.info(`dsh-context: projection rows backfilled for ${folded} session(s)`)
       if (paused) {

tests/host/backfill-sliced.spec.ts

@@ -0,0 +1,362 @@
+// The sliced cold fold of the projection warm-up (src/host/backfill.ts, issue #145) over the
+// REAL registry + the plugin's REAL units — byte-equality against one single-shot restore.
+
+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 { watchActivityBackfill } from '../../src/host/backfill'
+import { createContextTimelineDefinition } from '../../src/host/timeline'
+import { createContextActivityDefinition } from '../../src/host/activity'
+import { createContextHeadersDefinition } from '../../src/host/headers'
+import { arm, captureLogs, connectionOf, fakeState, trigger, until } from './helpers/backfill'
+import type { RouteBox } from './helpers/backfill'
+
+
+/** 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')
+  })
+
+  test('pacing rests proportionally to measured work and takes a logged recovery pause at the round budget', async () => {
+    const ctx = new Context()
+    const state = fakeState([1, 2, 3, 4].map(i => ({ header: { id: `s${i}`, cwd: `/repo/s${i}` } })))
+    const route = arm(ctx, state)
+    const lines = captureLogs(ctx)
+    const sleeps: number[] = []
+    let clock = 0
+    const dispose = watchActivityBackfill(ctx, undefined, {
+      now: () => (clock += 400), // every session measures 400ms of work
+      sleep: async (ms) => { sleeps.push(ms) },
+      minRestMs: 25,
+      workBudgetMs: 1_000,
+      roundPauseMs: 1_000,
+    })
+    await trigger(route)
+    await until(() => (state.coldSnapshots.length === 4 ? true : undefined), 'all four sessions folded')
+    await new Promise(resolve => setTimeout(resolve, 30))
+    dispose()
+    assert.deepEqual(state.coldSnapshots.sort(), ['s1', 's2', 's3', 's4'])
+    // Rests follow the measured 400ms cost; the third rest trips the 1000ms budget into a full pause.
+    assert.deepEqual(sleeps, [400, 400, 1_000, 400])
+    const pauses = lines.filter(l => l.level === 'info' && l.text.includes('pausing after'))
+    assert.equal(pauses.length, 1, 'the budget pause reports once')
+    assert.match(pauses[0].text, /pausing after 1s of fold work; resuming in 1s/)
+  })
+
+  test('sliced folds rest proportionally between slices, billed once across both granularities', async () => {
+    const ctx = new Context()
+    const rig = armSliced(ctx, new Map([['monster', fakeEvents(4_001)]]) )
+    const sleeps: number[] = []
+    let clock = 0
+    const dispose = watchActivityBackfill(ctx, undefined, {
+      now: () => (clock += 400),
+      sleep: async (ms) => { sleeps.push(ms) },
+      minRestMs: 25,
+      workBudgetMs: 60_000,
+      roundPauseMs: 2_000,
+    })
+    await trigger(rig.route)
+    await until(() => (rig.writes.has('monster') ? true : undefined), 'the write-back landed')
+    dispose()
+    // Two inter-slice rests (3 slices, 400ms per injected clock jump); the session-level rest
+    // covers only the non-sliced remainder (whole session 2800ms minus the 1200ms of slice work).
+    assert.deepEqual(sleeps, [400, 400, 1_600])
+  })
+
+  test('a zero-cost session rests exactly the floor', async () => {
+    const ctx = new Context()
+    const state = fakeState([{ header: { id: 'a', cwd: '/repo/a' } }, { header: { id: 'b', cwd: '/repo/b' } }])
+    const route = arm(ctx, state)
+    const sleeps: number[] = []
+    const dispose = watchActivityBackfill(ctx, undefined, {
+      now: () => 0,
+      sleep: async (ms) => { sleeps.push(ms) },
+      minRestMs: 25,
+      workBudgetMs: 60_000,
+      roundPauseMs: 2_000,
+    })
+    await trigger(route)
+    await until(() => (state.coldSnapshots.length === 2 ? true : undefined), 'both sessions folded')
+    dispose()
+    assert.deepEqual(sleeps, [25, 25])
+  })
+
+  test('dispose wakes a pending rest early (the pass never waits out a pause)', async () => {
+    const ctx = new Context()
+    const state = fakeState([{ header: { id: 'a', cwd: '/repo/a' } }, { header: { id: 'b', cwd: '/repo/b' } }])
+    const route = arm(ctx, state)
+    let sleeping = false
+    const dispose = watchActivityBackfill(ctx, undefined, {
+      now: () => 0,
+      sleep: async () => { sleeping = true; await new Promise(() => {}) }, // never resolves on its own
+      minRestMs: 25,
+      workBudgetMs: 60_000,
+      roundPauseMs: 2_000,
+    })
+    await trigger(route)
+    await until(() => (sleeping ? true : undefined), 'the first rest is pending')
+    dispose()
+    await new Promise(resolve => setTimeout(resolve, 30))
+    assert.deepEqual(state.coldSnapshots, ['a'], 'the abort woke the pending rest; the pass stopped without hanging')
+  })
+
+  test('a rest armed after dispose returns at once', async () => {
+    const ctx = new Context()
+    const state = fakeState([{ header: { id: 'a', cwd: '/repo/a' } }, { header: { id: 'b', cwd: '/repo/b' } }])
+    const route = arm(ctx, state)
+    let dispose: () => void = () => {}
+    let folds = 0
+    const gate = {
+      admit: <T>(read: () => Promise<T>): Promise<T | undefined> => {
+        const p = read()
+        folds++
+        if (folds === 2) dispose() // the second session's fold completes with the pass already disposed
+        return p
+      },
+    }
+    const sleeps: number[] = []
+    dispose = watchActivityBackfill(ctx, gate, {
+      now: () => 0,
+      sleep: async (ms) => { sleeps.push(ms) },
+      minRestMs: 25,
+      workBudgetMs: 60_000,
+      roundPauseMs: 2_000,
+    })
+    await trigger(route)
+    await until(() => (state.coldSnapshots.length === 2 ? true : undefined), 'both sessions folded')
+    await new Promise(resolve => setTimeout(resolve, 30))
+    // The post-dispose session-level rest skipped its sleep entirely, so only the first rest slept.
+    assert.deepEqual(sleeps, [25])
+  })
+
+  test('a budget trip inside one monster session pauses mid-fold and still completes the exact rows', async () => {
+    const ctx = new Context()
+    const events = fakeEvents(4_001)
+    const rig = armSliced(ctx, new Map([['monster', events]]))
+    const lines = captureLogs(ctx)
+    const sleeps: number[] = []
+    let clock = 0
+    const dispose = watchActivityBackfill(ctx, undefined, {
+      now: () => (clock += 400),
+      sleep: async (ms) => { sleeps.push(ms) },
+      minRestMs: 25,
+      workBudgetMs: 800,
+      roundPauseMs: 2_000,
+    })
+    await trigger(rig.route)
+    await until(() => (rig.writes.has('monster') ? true : undefined), 'the write-back landed')
+    dispose()
+    // Slice 1 rests 400; slice 2's rest trips the 800ms budget into a 2000ms pause; the
+    // session-level remainder (1600) trips it again. The pass resumes after each pause.
+    assert.deepEqual(sleeps, [400, 2_000, 2_000])
+    const pauses = lines.filter(l => l.level === 'info' && l.text.includes('pausing after'))
+    assert.equal(pauses.length, 2)
+    assert.match(pauses[0].text, /pausing after 1s of fold work; resuming in 2s/)
+    assert.deepStrictEqual(rig.writes.get('monster'), referenceRows(events, 'monster'), 'pauses never change the rows')
+  })
+})

tests/host/backfill.spec.ts

@@ -3,133 +3,9 @@
 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++) {
-    const value = read()
-    if (value !== undefined) return value
-    await new Promise(resolve => setTimeout(resolve, 1))
-  }
-  assert.fail(message)
-}
-
-interface RouteBox {
-  path?: string
-  fetch?: (request: Request) => Promise<Response>
-}
-
-/** Fire the captured trigger route (the client's dashboard-open POST). */
-async function trigger(route: RouteBox): Promise<Response> {
-  // The route half registers on cordis's deferred inject — settle first.
-  for (let i = 0; i < 300 && route.fetch === undefined; i++) {
-    await new Promise(resolve => setTimeout(resolve, 1))
-  }
-  assert.ok(route.fetch !== undefined, 'the backfill route registered')
-  return route.fetch(new Request(`http://dsh.test${BACKFILL_ROUTE}`, { method: 'POST' }))
-}
-
-interface LogLine {
-  level: string
-  text: string
-}
-
-function captureLogs(ctx: Context): LogLine[] {
-  const lines: LogLine[] = []
-  // Lift every source's threshold to debug: the default exporter drops warn/debug records before they reach the sink.
-  ctx.logger.exporter({
-    levels: { root: 3, 'dsh-context': 3 },
-    export(message) {
-      lines.push({ level: message.type, text: message.args.map(a => String(a)).join(' ') })
-    },
-  })
-  return lines
-}
-
-/** One format-refusal face of the persistence seam. The format edge throws
- * `SessionFormatUnsupportedMigrationError`, but the seam translates it — with
- * the "source vN artifact remains unchanged (raw log: …)" suffix — into
- * `SessionFormatUnsupportedError` before the error escapes (issue #75): the pass must classify BOTH faces as the same refusal.
- */
-function unsupportedFormatError(name: string): Error {
-  const error = new Error(
-    'subagent/descriptor 0 uses unsupported descriptor version 2'
-    + '; source v0 artifact remains unchanged (raw log: C:\\Users\\u\\.dsh\\sessions\\log.jsonl.zstd)',
-  )
-  error.name = name
-  return error
-}
-
-/** A connection face whose route registration lands in `route`. */
-function connectionOf(route: RouteBox): unknown {
-  return {
-    fetch: {
-      register: (r: { path: string; fetch: RouteBox['fetch'] }) => {
-        route.path = r.path
-        route.fetch = r.fetch
-        return () => {}
-      },
-    },
-  }
-}
-
-interface FakeState {
-  listed: unknown
-  served: Set<string>
-  live: Set<string>
-  coldReads: string[]
-  coldSnapshots: string[]
-  failReads: Map<string, unknown>
-}
-
-function arm(ctx: Context, state: FakeState): RouteBox {
-  const route: RouteBox = {}
-  ctx.provide('connection', connectionOf(route))
-  ctx.provide('sessionQuery', {
-    listSessions: async () => state.listed,
-  })
-  ctx.provide('sessionProjectionCache', {
-    cachedSnapshot: (header: { id: string }) =>
-      state.served.has(header.id)
-        ? { asOfSeq: 0, values: { contextActivity: { days: {} }, contextTimeline: { ok: true } } }
-        : undefined,
-    coldSnapshot: (header: { id: string }) => {
-      state.coldSnapshots.push(header.id)
-      return { asOfSeq: 0, values: {} }
-    },
-  })
-  ctx.provide('sessionPersistence', {
-    open: async (id: string) => {
-      state.coldReads.push(id)
-      if (state.failReads.has(id)) throw state.failReads.get(id)
-      const events: SessionEvent[] = []
-      return {
-        header: { id, version: 1, createdAt: 1, cwd: '/repo', isSeeded: false },
-        inheritedEventCount: 0,
-        read: async () => ({ events }),
-        close: async () => {},
-      }
-    },
-  })
-  ctx.provide('sessions', { get: (id: string) => (state.live.has(id) ? { id } : undefined) })
-  return route
-}
-
-function fakeState(listed: unknown): FakeState {
-  return {
-    listed,
-    served: new Set(),
-    live: new Set(),
-    coldReads: [],
-    coldSnapshots: [],
-    failReads: new Map(),
-  }
-}
+import { arm, captureLogs, connectionOf, fakeState, trigger, unsupportedFormatError, until } from './helpers/backfill'
+import type { RouteBox } from './helpers/backfill'
 
 describe('watchActivityBackfill', () => {
   test('a deployment without any service never fires either half', async () => {
@@ -766,211 +642,3 @@ describe('watchActivityBackfill', () => {
   })
 })
 
-/** 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')
-  })
-})

tests/host/helpers/backfill.ts

@@ -0,0 +1,128 @@
+// Shared rig for the backfill specs (tests/host/backfill.spec.ts and backfill-sliced.spec.ts):
+// the trigger route box, the log capture, and the fake cold-path faces.
+
+import assert from 'node:assert/strict'
+import type { Context } from '@deepseek-ai/cordis'
+import type { SessionEvent } from '@deepseek-ai/dsh-session'
+import { BACKFILL_ROUTE } from '../../../src/host/backfill'
+
+export async function until<T>(read: () => T | undefined, message: string): Promise<T> {
+  for (let i = 0; i < 300; i++) {
+    const value = read()
+    if (value !== undefined) return value
+    await new Promise(resolve => setTimeout(resolve, 1))
+  }
+  assert.fail(message)
+}
+
+export interface RouteBox {
+  path?: string
+  fetch?: (request: Request) => Promise<Response>
+}
+
+/** Fire the captured trigger route (the client's dashboard-open POST). */
+export async function trigger(route: RouteBox): Promise<Response> {
+  // The route half registers on cordis's deferred inject — settle first.
+  for (let i = 0; i < 300 && route.fetch === undefined; i++) {
+    await new Promise(resolve => setTimeout(resolve, 1))
+  }
+  assert.ok(route.fetch !== undefined, 'the backfill route registered')
+  return route.fetch(new Request(`http://dsh.test${BACKFILL_ROUTE}`, { method: 'POST' }))
+}
+
+export interface LogLine {
+  level: string
+  text: string
+}
+
+export function captureLogs(ctx: Context): LogLine[] {
+  const lines: LogLine[] = []
+  // Lift every source's threshold to debug: the default exporter drops warn/debug records before they reach the sink.
+  ctx.logger.exporter({
+    levels: { root: 3, 'dsh-context': 3 },
+    export(message) {
+      lines.push({ level: message.type, text: message.args.map(a => String(a)).join(' ') })
+    },
+  })
+  return lines
+}
+
+/** One format-refusal face of the persistence seam. The format edge throws
+ * `SessionFormatUnsupportedMigrationError`, but the seam translates it — with
+ * the "source vN artifact remains unchanged (raw log: …)" suffix — into
+ * `SessionFormatUnsupportedError` before the error escapes (issue #75): the pass must classify BOTH faces as the same refusal.
+ */
+export function unsupportedFormatError(name: string): Error {
+  const error = new Error(
+    'subagent/descriptor 0 uses unsupported descriptor version 2'
+    + '; source v0 artifact remains unchanged (raw log: C:\\Users\\u\\.dsh\\sessions\\log.jsonl.zstd)',
+  )
+  error.name = name
+  return error
+}
+
+/** A connection face whose route registration lands in `route`. */
+export function connectionOf(route: RouteBox): unknown {
+  return {
+    fetch: {
+      register: (r: { path: string; fetch: RouteBox['fetch'] }) => {
+        route.path = r.path
+        route.fetch = r.fetch
+        return () => {}
+      },
+    },
+  }
+}
+
+export interface FakeState {
+  listed: unknown
+  served: Set<string>
+  live: Set<string>
+  coldReads: string[]
+  coldSnapshots: string[]
+  failReads: Map<string, unknown>
+}
+
+export function arm(ctx: Context, state: FakeState): RouteBox {
+  const route: RouteBox = {}
+  ctx.provide('connection', connectionOf(route))
+  ctx.provide('sessionQuery', {
+    listSessions: async () => state.listed,
+  })
+  ctx.provide('sessionProjectionCache', {
+    cachedSnapshot: (header: { id: string }) =>
+      state.served.has(header.id)
+        ? { asOfSeq: 0, values: { contextActivity: { days: {} }, contextTimeline: { ok: true } } }
+        : undefined,
+    coldSnapshot: (header: { id: string }) => {
+      state.coldSnapshots.push(header.id)
+      return { asOfSeq: 0, values: {} }
+    },
+  })
+  ctx.provide('sessionPersistence', {
+    open: async (id: string) => {
+      state.coldReads.push(id)
+      if (state.failReads.has(id)) throw state.failReads.get(id)
+      const events: SessionEvent[] = []
+      return {
+        header: { id, version: 1, createdAt: 1, cwd: '/repo', isSeeded: false },
+        inheritedEventCount: 0,
+        read: async () => ({ events }),
+        close: async () => {},
+      }
+    },
+  })
+  ctx.provide('sessions', { get: (id: string) => (state.live.has(id) ? { id } : undefined) })
+  return route
+}
+
+export function fakeState(listed: unknown): FakeState {
+  return {
+    listed,
+    served: new Set(),
+    live: new Set(),
+    coldReads: [],
+    coldSnapshots: [],
+    failReads: new Map(),
+  }
+}