diff --git a/src/host/backfill.ts b/src/host/backfill.ts index cad6eb3..95da649 100644 --- a/src/host/backfill.ts +++ b/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 } -/** 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 + 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 | 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 | null, log: { header: SessionHeader; inheritedEventCount: unknown; events: readonly SessionEvent[] }, signal: AbortSignal, + rest: (costMs: number) => Promise, + now: () => number, ): Promise { 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 => { + if (abort.signal.aborted) return + // Assigned synchronously by the Promise executor below. + let onAbort!: () => void + try { + await Promise.race([ + pacing.sleep(ms), + new Promise((resolve) => { + onAbort = () => { resolve() } + abort.signal.addEventListener('abort', onAbort, { once: true }) + }), + ]) + } finally { + abort.signal.removeEventListener('abort', onAbort) + } + } + const rest = async (costMs: number): Promise => { + 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) { diff --git a/tests/host/backfill-sliced.spec.ts b/tests/host/backfill-sliced.spec.ts new file mode 100644 index 0000000..4a5b74b --- /dev/null +++ b/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 + 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, opts: { + liveOnPrepare?: Set + failPrepare?: Set + 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) => { + 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() + const sessionsFace: Record = { + 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: (read: () => Promise): Promise => { + 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') + }) +}) diff --git a/tests/host/backfill.spec.ts b/tests/host/backfill.spec.ts index 7ea5d58..9548da3 100644 --- a/tests/host/backfill.spec.ts +++ b/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(read: () => T | undefined, message: string): Promise { - 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 -} - -/** Fire the captured trigger route (the client's dashboard-open POST). */ -async function trigger(route: RouteBox): Promise { - // 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 - live: Set - coldReads: string[] - coldSnapshots: string[] - failReads: Map -} - -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 - 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, opts: { - liveOnPrepare?: Set - failPrepare?: Set - 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) => { - 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() - const sessionsFace: Record = { - 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') - }) -}) diff --git a/tests/host/helpers/backfill.ts b/tests/host/helpers/backfill.ts new file mode 100644 index 0000000..e5f5ec2 --- /dev/null +++ b/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(read: () => T | undefined, message: string): Promise { + 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 +} + +/** Fire the captured trigger route (the client's dashboard-open POST). */ +export async function trigger(route: RouteBox): Promise { + // 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 + live: Set + coldReads: string[] + coldSnapshots: string[] + failReads: Map +} + +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(), + } +}