diff --git a/docs/compatibility.md b/docs/compatibility.md index dad8c46..ce17af1 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -111,12 +111,14 @@ These checks prove source-level seam compatibility, statistical parity with the The skill-injection category (issue #66) bumps the timeline projection's `stateVersion` (18 → 19): skill content — the `` catalog digest, a user-explicit `/name` invocation's instructions message, and the content a `skill`-tool load returns — now folds into its own `skill` composition bucket instead of `inject`/`tool`. The re-bucketing changes the fold's per-category sums, so on upgrade every cached per-session row is invalidated and re-folded from the durable log, which rebuilds them under the new categories. The bump orphans the `contextTimeline` key for idle sessions, which have no refresh channel until they go live again — such a session re-folds when its log receives its next event. New sessions are exact from their first event. -The Agent network card covers the orphaned relatives without waiting for a visit: a session listed without a `contextTimeline` row (composition absent, occupancy only) fetches its slim head from the plugin's `/api/dsh-context/detail` fetch route (mounted through Connection's exact-route registry, behind the same authenticated `/api` fence), which folds the durable log on demand — the node's ring renders the full composition the first time the card opens. The same backfill serves sessions whose cache predates the plugin's installation entirely. Where the route is unavailable (the baseline gate, a connection service without the exact-route registry), the card degrades to the pressure-only occupancy ring as before. +The Agent network card covers the orphaned relatives without waiting for a visit: a session listed without a `contextTimeline` row (composition absent, occupancy only) fetches its slim head from the plugin's `/api/dsh-context/detail` fetch route (mounted through Connection's exact-route registry, behind the same authenticated `/api` fence), which folds the durable log on demand — the node's ring renders the full composition the first time the card opens. The same backfill serves sessions whose cache predates the plugin's installation entirely. Where the route is unavailable (the baseline gate, a connection service without the exact-route registry) — or the cold-read gate below skips the read under heap pressure — the card degrades to the pressure-only occupancy ring as before. The session cards' last-user-message preview (`lastUser` on the `contextTimeline` head) bumps the timeline projection's `stateVersion` (19 → 20): a stale row can never gain the field while its session is idle (it only lands when a user message folds), so on upgrade every cached timeline row is invalidated and refolded from the durable log. To keep the bump from orphaning idle sessions' rows, the warm-up (`src/host/backfill.ts`) probes BOTH insight rows — `contextActivity` AND `contextTimeline` — and cold-refolds every session whose either row is missing or version-stale, so previews (and every other head figure) are exact for the whole session list the first time the page opens after the upgrade. A deployment without the cold-path services — or whose client cannot mount the warm-up's trigger route — degrades to the per-session refold on the session's next activity, the same as pre-warm-up upgrades. The Context Insights page (the sidebar panel list's first-level `main` panel) reads its `contextActivity` projection — a per-day ledger of billed tokens and completed requests, `stateVersion` 1 — off every session-list row's projection column. A new key is purely additive: no existing cached row is invalidated, but sessions folded before the unit existed (and sessions that predate the plugin entirely, whose `contextTimeline`/`contextHeaders` rows are missing too) have nothing to serve until they next go live. The plugin therefore runs a one-pass warm-up on demand (`src/host/backfill.ts`): the page is the rows' only reader, so it POSTs the plugin's `/api/dsh-context/backfill` fetch route (the same authenticated `/api` fence as the detail route) the first time it mounts, and only then does the host walk the corpus — for every stored session missing the row it cold-reads the durable log once through the projection cache's own cold-read ladder (`coldSnapshot`), which seeds each unit from its cached rows, folds the remainder, and writes the refreshed checkpoint back. The pass runs once per host process (later opens and later POSTs are no-ops); live sessions are skipped (they fold for themselves); a legacy log the running harness refuses to migrate is permanent — its per-session detail logs at debug and the pass ends with one summary info line, so an unmigratable corpus costs one line per page-open process, not one warn per session per boot; and a deployment whose client never opens the page never cold-reads a thing. A deployment without the sessionQuery/sessionProjectionCache/sessionPersistence/sessions services — or whose connection service lacks the exact-route registry — simply never arms the pass (such a client could not open the page to read the rows anyway). +Every cold read the plugin makes — this pass's per-session fold and the detail route's cold rung alike — runs through one host-wide cold-read gate (`src/host/coldRead.ts`, issue #121): reads run one at a time (FIFO), and a read starts only while the process's V8 heap sits below two-thirds of its own ceiling (the persistence backend decodes a whole log per read, so an unbounded fan-out of 10k-event sessions stacks several-hundred-MB object graphs on a heap whose ceiling may not move — a capped Electron main). A read the gate skips under pressure takes the caller's designed no-data path rather than failing: the pass stops early with one info line (folded rows persist; the rest refold on their next live activity or a later process's pass), and the detail route answers its typed null. + The page's per-day usage chart (the first row's right half) extends the `contextActivity` ledger with per-day pricing records — each day's billed buckets keyed provider → model → pricing period, the same `SessionCostUsage` raw material the timeline's session-cost totals carry, booked at each turn's initiation instant and priced client-side by the same book and estimator as the KPI band. Booking fees per (provider, model, period) changes the fold's state shape, so the unit's `stateVersion` bumps (1 → 2): on upgrade every cached `contextActivity` row reads version-stale, and the page's existing warm-up (above) cold-refolds the corpus from the durable logs, rebuilding every day's tokens AND fees exactly the first time the panel opens. On the wire the field is additive-optional: a day folded before it existed (or whose settlements carried no model header or no usage) serves without it, and the chart degrades to tokens-only bars for those days — the same sanitizer contract every other projection value rides. The timing card's session-time strip (the card's bottom band: every completed step's TTFT wait, decode blocks in stream order, tool-run windows, and in-step residue packed gapless in occurrence order — each slice's width is its share of the session's cumulative active time, with ticks at zero, the quartiles, and the full span) bumps the timeline projection's `stateVersion` (22 → 23): the strip's painted spans are a per-step collection the fold flushes at `step/end`, which later events cannot backfill for steps already folded (the timing totals' own v11 precedent), so on upgrade every cached timeline row is invalidated and refolded from the durable log, which rebuilds the spans for sessions started under older plugin builds; the v20 warm-up rebuilds idle sessions' rows the first time the page opens. On the wire the collection is additive-optional — it rides the detail channel (and the inline value on channel-less hosts), never the slim head — so a payload folded before it existed serves without it, and the timing card shows no strip then. The spans are bounded to the newest 2,000 records (a plain count cap), so an outsize session's strip paints its recent window; untrusted spans are re-proved per item at the client's render boundary, the same sanitizer contract every collection rides. diff --git a/src/host/backfill.ts b/src/host/backfill.ts index c961654..ad98bdb 100644 --- a/src/host/backfill.ts +++ b/src/host/backfill.ts @@ -46,6 +46,7 @@ import type { Context } from '@deepseek-ai/cordis' import { interruptedTurnClosers, SessionLogOffset } from '@deepseek-ai/dsh-session' import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session' +import { type ColdReadGate, makeColdReadGate } from './coldRead' /** The plugin's warm-up trigger route, under the authenticated `/api` fence. */ export const BACKFILL_ROUTE = '/api/dsh-context/backfill' @@ -232,9 +233,12 @@ function messageOf(error: unknown): string { * when the dashboard first opens; the cold-path faces compose `launch`. * Either half may land first; the first request after both launches the one * pass, and every later request is a no-op. Returns the deferred injects' - * disposer (abort on unload). + * disposer (abort on unload). Every cold read runs through the shared + * host-wide gate (coldRead.ts): one read at a time, and the pass stops + * early when the gate skips under heap pressure — folded rows persist and + * every remaining session keeps its on-demand refold paths. */ -export function watchActivityBackfill(ctx: Context): () => void { +export function watchActivityBackfill(ctx: Context, coldReads: ColdReadGate = makeColdReadGate()): () => void { let requested = false let started = false let launch: (() => void) | null = null @@ -292,6 +296,7 @@ export function watchActivityBackfill(ctx: Context): () => void { if (!Array.isArray(listed)) return let folded = 0 let refused = 0 + let paused = false for (const record of listed) { if (abort.signal.aborted) return const header = headerOf(record) @@ -299,10 +304,20 @@ export function watchActivityBackfill(ctx: Context): () => void { if (isLive(sessions, header.id)) continue if (servesRows(cache as unknown as ProjectionCacheLike, header)) continue try { - 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 only the probe's identity witness. - ;(cache as unknown as ProjectionCacheLike).coldSnapshot(log.header, log.inheritedEventCount, log.events) + 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 only the probe's identity witness. + ;(cache as unknown as ProjectionCacheLike).coldSnapshot(log.header, log.inheritedEventCount, log.events) + return true + }) + if (admitted === undefined) { + // Skipped under heap pressure: end the pass here — pressure does + // not fall within a pacing tick, so hammering the gate per + // remaining session only delays the host's own work. + paused = true + break + } folded++ } catch (error: unknown) { if (isUnsupportedFormat(error)) { @@ -318,6 +333,9 @@ export function watchActivityBackfill(ctx: Context): () => void { await new Promise(resolve => setTimeout(resolve, YIELD_MS)) } if (folded > 0) ctx.logger.info(`dsh-context: projection rows backfilled for ${folded} session(s)`) + if (paused) { + ctx.logger.info('dsh-context: projection backfill stopped early under heap pressure (remaining sessions refold on their next live activity)') + } if (refused > 0) { ctx.logger.info(`dsh-context: projection backfill skipped ${refused} session(s) (legacy log format is not migratable; sources left unchanged)`) } diff --git a/src/host/coldRead.ts b/src/host/coldRead.ts new file mode 100644 index 0000000..f13d966 --- /dev/null +++ b/src/host/coldRead.ts @@ -0,0 +1,86 @@ +/** + * The host-wide cold-read governor (issue #121). + * + * Every cold read — the warm-up pass's per-session fold (backfill.ts) and + * the detail route's cold rung (detail.ts) — materializes one session's + * whole durable log: the persistence backend decodes every frame before it + * can slice a single event, so a 10k-event session lands as a several- + * hundred-MB object graph, and the harness pins its own floors on top (the + * backend's two-entry decoded-log memo, the observation reader's five-entry + * prepared-session cache). Those floors are the harness's to tune; what the + * PLUGIN owns is how many of its cold reads run at once and whether a new + * one starts on an already-hot heap. Before this gate the answer to both + * was "unbounded": the dashboard's first open fires the corpus pass and the + * session cards' / Agent network's per-session detail fan-out together, and + * on a capped-heap host (an Electron main whose V8 ceiling cannot move) + * that stacking was enough to push the process over its limit. + * + * The gate therefore admits one cold read at a time (FIFO) and only while + * the heap leaves headroom for it. A skipped read never fails visibly: the + * pass stops early (its folded rows persist; the rest refold on their next + * live activity or a later process's pass) and the detail route answers its + * typed null (the client keeps its last detail, the cards their no-data + * note) — every degradation already designed, no functionality removed. + */ + +import { getHeapStatistics } from 'node:v8' + +/** One heap-headroom reading: bytes in use against the process's own V8 ceiling. */ +export interface HeapReading { + used: number + limit: number +} + +/** The admission face both cold paths share. */ +export interface ColdReadGate { + /** + * Run one cold read once every earlier read has settled, when the heap + * leaves headroom for it; `undefined` when the read was skipped under + * pressure — the caller's own no-data path answers instead. + */ + admit(read: () => Promise): Promise +} + +/** + * The high-water mark as a fraction of the process's own heap ceiling: a + * cold read starts only below two-thirds full, leaving the top third for + * the read itself plus the harness's floors. The ceiling is V8's real + * limit, so every host class — a default-heap CLI, a capped Electron + * main — gets its own mark from the same constant. + */ +const HIGH_WATER = 2 / 3 + +/** The process's own V8 reading. */ +function v8Reading(): HeapReading { + const stats = getHeapStatistics() + return { used: stats.used_heap_size, limit: stats.heap_size_limit } +} + +const noop = (): void => {} + +/** + * One governor per host process, shared by both cold paths. `reading` is + * injectable for tests; a throwing or garbage probe fails OPEN — the meter + * must never block a read (the version probe's philosophy). + */ +export function makeColdReadGate(reading: () => HeapReading = v8Reading): ColdReadGate { + const clear = (): boolean => { + try { + const { used, limit } = reading() + if (!Number.isFinite(used) || !Number.isFinite(limit) || limit <= 0) return true + return used < limit * HIGH_WATER + } catch { + return true + } + } + let tail: Promise = Promise.resolve() + return { + admit(read: () => Promise): Promise { + // The headroom check runs when the read reaches the front of the + // queue, not when it was queued — pressure changes while waiting. + const run = tail.then(() => (clear() ? read() : undefined)) + tail = run.then(noop, noop) + return run + }, + } +} diff --git a/src/host/detail.ts b/src/host/detail.ts index 925e229..e266edb 100644 --- a/src/host/detail.ts +++ b/src/host/detail.ts @@ -37,6 +37,7 @@ import type { Context } from '@deepseek-ai/cordis' import type { FoldBounds } from './config' +import { type ColdReadGate, makeColdReadGate } from './coldRead' import { applyTimeline, buildTimelineDetail, createTimelineState } from './fold' /** The plugin's detail route, under the authenticated `/api` fence. */ @@ -90,7 +91,7 @@ function failure(code: string, message: string): Response { * registration rides the injected fiber: either service unloading withdraws * the route and closes the gate. */ -export function watchDetailChannel(ctx: Context, bounds: FoldBounds): DetailChannelGate { +export function watchDetailChannel(ctx: Context, bounds: FoldBounds, coldReads: ColdReadGate = makeColdReadGate()): DetailChannelGate { const gate = { live: false } ctx.inject(['connection', 'sessions'], (c) => { const connection = c.get('connection') as ConnectionHostFace | undefined @@ -131,24 +132,30 @@ export function watchDetailChannel(ctx: Context, bounds: FoldBounds): DetailChan // Cold session (viewed through a prepared observation, never entered // into the live store): observe it and fold the detail from its // immutable log. The lease disposes promptly; the query's prepared - // cache retains the session for reuse. A session nothing can observe - // resolves to the typed null — the client keeps its last detail. + // cache retains the session for reuse. The read runs through the + // shared cold-read gate (coldRead.ts): one log decode at a time + // host-wide, and a skip under heap pressure resolves to the same + // typed null as a session nothing can observe — the client keeps + // its last detail either way. const query = ctx.get('sessionQuery') as SessionQueryFace | undefined const observe = typeof query?.observeSession === 'function' ? query.observeSession.bind(query) : undefined if (observe === undefined) return reply({ ok: true, value: null }) - const observation = await observe(sessionId, { projectionMode: 'none' }) - const events = (observation as { events?: unknown } | null)?.events - if (!Array.isArray(events)) return reply({ ok: true, value: null }) - let state = createTimelineState() - try { - for (const ev of events) state = applyTimeline(state, ev as never, bounds) - } finally { - const dispose = (observation as { [Symbol.dispose]?: unknown } | null)?.[Symbol.dispose] - if (typeof dispose === 'function') dispose.call(observation) - } - return reply({ ok: true, value: buildTimelineDetail(state, bounds) }) + const value = await coldReads.admit(async () => { + const observation = await observe(sessionId, { projectionMode: 'none' }) + const events = (observation as { events?: unknown } | null)?.events + if (!Array.isArray(events)) return null + let state = createTimelineState() + try { + for (const ev of events) state = applyTimeline(state, ev as never, bounds) + } finally { + const dispose = (observation as { [Symbol.dispose]?: unknown } | null)?.[Symbol.dispose] + if (typeof dispose === 'function') dispose.call(observation) + } + return buildTimelineDetail(state, bounds) + }) + return reply({ ok: true, value: value ?? null }) } catch (err) { return failure('gateway/internal', err instanceof Error ? err.message : String(err)) } diff --git a/src/host/index.ts b/src/host/index.ts index 8893924..c6cc1a2 100644 --- a/src/host/index.ts +++ b/src/host/index.ts @@ -25,6 +25,7 @@ import { createContextActivityDefinition } from './activity' import { createToolAttribution } from './attribution' import { watchBalanceChannel } from './balance' import { watchActivityBackfill } from './backfill' +import { makeColdReadGate } from './coldRead' import { Config, resolveBounds } from './config' import { watchDetailChannel } from './detail' import { createFallbackActivityDefinition, createFallbackHeadersDefinition, createFallbackTimelineDefinition } from './fallback' @@ -76,11 +77,16 @@ export function apply(ctx: Context, config: Config): void { // for any pre-step message that would persist unidentified — the harness's // load path refuses such events wholesale, permanently bricking the session. watchStepIdentity(ctx) + // The host-wide cold-read governor (coldRead.ts, issue #121): the detail + // route's cold rung and the overview warm-up share one FIFO + heap + // high-water mark, so the plugin's own log decodes never stack on a hot + // heap — each skipped read takes its caller's designed no-data path. + const coldReads = makeColdReadGate() // The split wire generation (detail.ts): the detail channel arms whenever // the connection/sessions/webServer services compose (load order never // assumed), and the unit's view reads the gate per serve — slim while the // channel is live, inline otherwise. - const gate = watchDetailChannel(ctx, resolveBounds(config)) + const gate = watchDetailChannel(ctx, resolveBounds(config), coldReads) ctx.sessionProjections.register(createContextTimelineDefinition(config, () => gate.live)) ctx.sessionProjections.register(createContextHeadersDefinition(name => attribution.ownerOf(name))) ctx.sessionProjections.register(createContextActivityDefinition()) @@ -88,7 +94,7 @@ export function apply(ctx: Context, config: Config): void { // a unit existed get their rows from one background cold read each, run on // demand — the dashboard (the rows' only reader) summons the pass through // the plugin's fetch route the first time it opens. - watchActivityBackfill(ctx) + watchActivityBackfill(ctx, coldReads) installSettings(ctx) } diff --git a/tests/host/backfill.spec.ts b/tests/host/backfill.spec.ts index 433ec07..6beb34e 100644 --- a/tests/host/backfill.spec.ts +++ b/tests/host/backfill.spec.ts @@ -739,4 +739,46 @@ describe('watchActivityBackfill', () => { await new Promise(resolve => setTimeout(resolve, 30)) dispose() }) + + test('every cold read runs through the shared gate, and a skip stops the pass early', async () => { + const ctx = new Context() + const state = fakeState([ + { header: { id: 'a', cwd: '/repo/a' } }, + { header: { id: 'b', cwd: '/repo/b' } }, + { header: { id: 'c', cwd: '/repo/c' } }, + ]) + const route = arm(ctx, state) + const lines = captureLogs(ctx) + let admitted = 0 + const dispose = watchActivityBackfill(ctx, { + // The first read runs; every later one is skipped (the heap-pressure answer). + admit: (read: () => Promise): Promise => { + admitted++ + return admitted === 1 ? read() : Promise.resolve(undefined) + }, + }) + await trigger(route) + await until(() => (lines.some(l => l.text.includes('heap pressure')) ? true : undefined), 'the early-stop line') + await new Promise(resolve => setTimeout(resolve, 40)) + dispose() + assert.deepEqual(state.coldReads, ['a'], 'the pass stopped at the first skip') + assert.deepEqual(state.coldSnapshots, ['a'], 'the folded session kept its row') + assert.equal(admitted, 2, 'one run plus one skip — the loop did not hammer the gate') + assert.ok(lines.some(l => l.level === 'info' && l.text.includes('backfilled for 1 session(s)')), 'the folded count still reports') + assert.ok(lines.some(l => l.level === 'info' && l.text.includes('stopped early under heap pressure')), 'the early stop reports once') + }) + + test('a pass admitted under pressure from the start folds nothing and says so', async () => { + const ctx = new Context() + const state = fakeState([{ header: { id: 'a', cwd: '/repo/a' } }]) + const route = arm(ctx, state) + const lines = captureLogs(ctx) + const dispose = watchActivityBackfill(ctx, { admit: () => Promise.resolve(undefined) }) + await trigger(route) + await until(() => (lines.some(l => l.text.includes('heap pressure')) ? true : undefined), 'the early-stop line') + await new Promise(resolve => setTimeout(resolve, 20)) + dispose() + assert.deepEqual(state.coldReads, [], 'nothing was read') + assert.equal(lines.filter(l => l.text.includes('heap pressure')).length, 1, 'one line, not one per session') + }) }) diff --git a/tests/host/coldRead.spec.ts b/tests/host/coldRead.spec.ts new file mode 100644 index 0000000..0340be7 --- /dev/null +++ b/tests/host/coldRead.spec.ts @@ -0,0 +1,123 @@ +// The cold-read governor (src/host/coldRead.ts, issue #121): FIFO exclusion +// across concurrent admits, the two-thirds high-water skip checked when the +// read reaches the FRONT of the queue (not when queued), fail-open on a +// hostile or garbage probe, and queue survival across a rejecting read. + +import assert from 'node:assert/strict' +import { describe, test } from 'vitest' +import { type HeapReading, makeColdReadGate } from '../../src/host/coldRead' + +const LOW: HeapReading = { used: 100, limit: 1000 } +const HIGH: HeapReading = { used: 800, limit: 1000 } + +/** One manually-settled promise. */ +function deferred(): { promise: Promise; resolve: (value: T) => void } { + let resolve!: (value: T) => void + const promise = new Promise((r) => { resolve = r }) + return { promise, resolve } +} + +describe('makeColdReadGate', () => { + test('concurrent reads run one at a time, in queue order', async () => { + const gate = makeColdReadGate(() => LOW) + const order: string[] = [] + let active = 0 + let maxActive = 0 + const task = (name: string, gate2?: Promise): (() => Promise) => async () => { + active++ + maxActive = Math.max(maxActive, active) + order.push(`${name}:start`) + const value = await (gate2 ?? Promise.resolve(name)) + order.push(`${name}:end`) + active-- + return value + } + const blocker = deferred() + const first = gate.admit(task('a', blocker.promise)) + const second = gate.admit(task('b')) + const third = gate.admit(task('c')) + await new Promise(resolve => setTimeout(resolve, 10)) + assert.deepEqual(order, ['a:start'], 'the queued reads wait for the front') + blocker.resolve('A') + assert.deepEqual(await Promise.all([first, second, third]), ['A', 'b', 'c']) + assert.deepEqual(order, ['a:start', 'a:end', 'b:start', 'b:end', 'c:start', 'c:end']) + assert.equal(maxActive, 1, 'never two reads in flight') + }) + + test('a read over the high-water mark is skipped, not run', async () => { + const gate = makeColdReadGate(() => HIGH) + let ran = 0 + const value = await gate.admit(async () => { + ran++ + return 42 + }) + assert.equal(value, undefined) + assert.equal(ran, 0, 'the read never started') + }) + + test('the headroom check runs at the front of the queue, not at queue time', async () => { + let reading = LOW + const gate = makeColdReadGate(() => reading) + const blocker = deferred() + let started = false + const first = gate.admit(async () => { + started = true + const value = await blocker.promise + return value + }) + let ran = 0 + const second = gate.admit(async () => { + ran++ + return 'b' + }) + // Let the first read actually START (its check passed), then let pressure + // rise while the second read waits behind it. + await new Promise(resolve => setTimeout(resolve, 10)) + assert.ok(started, 'the front read started under the low reading') + reading = HIGH + blocker.resolve('a') + assert.equal(await first, 'a', 'the running read settles normally') + assert.equal(await second, undefined, 'the queued read meets the hot heap at its turn') + assert.equal(ran, 0) + // And pressure falling again admits the next read. + reading = LOW + assert.equal(await gate.admit(async () => 'c'), 'c') + }) + + test('the boundary is strict: exactly two-thirds full does not admit', async () => { + const at = makeColdReadGate(() => ({ used: 200, limit: 300 })) + assert.equal(await at.admit(async () => 'x'), undefined) + const below = makeColdReadGate(() => ({ used: 199, limit: 300 })) + assert.equal(await below.admit(async () => 'x'), 'x') + }) + + test('a hostile or garbage probe fails open', async () => { + const throwing = makeColdReadGate(() => { + throw new Error('hostile meter') + }) + assert.equal(await throwing.admit(async () => 'x'), 'x') + for (const junk of [ + { used: Number.NaN, limit: 1000 }, + { used: 100, limit: Number.NaN }, + { used: 100, limit: 0 }, + { used: 100, limit: -1 }, + ]) { + const gate = makeColdReadGate(() => junk) + assert.equal(await gate.admit(async () => 'x'), 'x', `junk reading ${JSON.stringify(junk)} admits`) + } + }) + + test('a rejecting read propagates and the queue survives it', async () => { + const gate = makeColdReadGate(() => LOW) + const failure = gate.admit(async (): Promise => { + throw new Error('fold exploded') + }) + await assert.rejects(failure, /fold exploded/) + assert.equal(await gate.admit(async () => 'after'), 'after', 'the next read still runs') + }) + + test('the default reading is the process’s own V8 heap', async () => { + const gate = makeColdReadGate() + assert.equal(await gate.admit(async () => 'x'), 'x') + }) +}) diff --git a/tests/host/detail.spec.ts b/tests/host/detail.spec.ts index de442b5..3ad7cd9 100644 --- a/tests/host/detail.spec.ts +++ b/tests/host/detail.spec.ts @@ -267,6 +267,42 @@ describe('the detail route', () => { assert.equal(result.error.message, 'persistence down') }) + test('a cold read skipped under heap pressure answers the typed null (and never observes)', async () => { + let observed = 0 + const { ctx, captured } = ctxOf({ + connection: { fetch: { register: () => () => {} } }, + sessions: sessionsWith(okSession()), + stateOf: () => undefined, + sessionQuery: { + observeSession: () => { + observed++ + return Promise.resolve({ events: [] }) + }, + }, + }) + let admitted = 0 + watchDetailChannel(ctx, BOUNDS, { + admit: (_read: () => Promise): Promise => { + admitted++ + return Promise.resolve(undefined) + }, + }) + assert.deepEqual(await call(captured, { sessionId: 'cold1' }), { ok: true, value: null }) + assert.equal(admitted, 1, 'the cold read went through the gate') + assert.equal(observed, 0, 'a skipped read never touches the log') + }) + + test('the live path never touches the cold-read gate', async () => { + const { state } = driveTimeline([userMessage(1, [{ type: 'text', text: 'hi' }], { kind: 'user' })]) + const { ctx, captured } = liveCtx(state) + watchDetailChannel(ctx, BOUNDS, { + admit: () => Promise.reject(new Error('the gate must not see a live read')), + }) + const result = await call(captured, { sessionId: 's1' }) as { ok: boolean; value: { rev: number } | null } + assert.equal(result.ok, true) + assert.ok(result.value !== null, 'the live read served its detail without the gate') + }) + test('serves the fold state\'s detail payload with its revision', async () => { const { state } = driveTimeline([ header(1, { model: 'deepseek-v4-flash', provider: 'deepseek' }),