From a95a2c5bf8f51d06e8c5fefe7b0d1a6abc7d0552 Mon Sep 17 00:00:00 2001 From: ozymandiashh <234437643+ozymandiashh@users.noreply.github.com> Date: Wed, 12 Aug 2026 20:31:17 +0300 Subject: [PATCH] fix(desktop): close cache and lifecycle review gaps --- app/electron/cli.test.ts | 219 ++++- app/electron/cli.ts | 135 +++- .../CodeBurnMenubar/Data/DataClient.swift | 54 +- .../Data/ServeConnection.swift | 489 ++++++++--- .../ServeConnectionTests.swift | 761 ++++++++++++++++-- src/codex-cache.ts | 53 +- src/parser.ts | 115 ++- src/providers/antigravity.ts | 71 +- src/serve.ts | 236 +++++- src/sync/ledger.ts | 100 ++- tests/cache-directory-switch.test.ts | 245 ++++++ tests/parser.test.ts | 159 +++- tests/providers/claude-config-dirs.test.ts | 31 + tests/serve-stdio.test.ts | 176 +++- tests/sync-ledger-otlp.test.ts | 168 +++- 15 files changed, 2642 insertions(+), 370 deletions(-) create mode 100644 tests/cache-directory-switch.test.ts diff --git a/app/electron/cli.test.ts b/app/electron/cli.test.ts index b9f90015..95c1d690 100644 --- a/app/electron/cli.test.ts +++ b/app/electron/cli.test.ts @@ -1,5 +1,5 @@ // @vitest-environment node -import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest' +import { describe, it, expect, beforeEach, afterEach } from 'vitest' import { mkdtempSync, mkdirSync, writeFileSync, readFileSync, rmSync, chmodSync } from 'node:fs' import { tmpdir } from 'node:os' import { dirname, join, isAbsolute, relative, win32, posix } from 'node:path' @@ -27,6 +27,14 @@ function readMaybe(path: string): string { try { return readFileSync(path, 'utf8') } catch { return '' } } +async function waitFor(condition: () => boolean, timeoutMs = 3_000): Promise { + const deadline = Date.now() + timeoutMs + while (!condition()) { + if (Date.now() > deadline) throw new Error('waitFor timed out') + await new Promise(resolve => setTimeout(resolve, 10)) + } +} + /** A protocol-faithful fake CLI whose serve child accepts requests before its * delayed ready frame. Files expose process starts and heavy request executions * without relying on timing or private ServeClient internals. */ @@ -405,19 +413,19 @@ describe('spawnCli coalescing (read-only)', () => { expect(readFileSync(countFile, 'utf8')).toBe('x') // exactly one spawn }) - it('spawns again once the 5s result cache has expired', async () => { - vi.useFakeTimers({ toFake: ['Date'] }) - try { - const countFile = join(dir, 'spawns') - fakeBin('counter-ttl.js', `require('fs').appendFileSync(${JSON.stringify(countFile)},'x'); process.stdout.write(JSON.stringify({ok:1}))`) - vi.setSystemTime(0) - await spawnCli(['status']) - vi.setSystemTime(6_000) - await spawnCli(['status']) - expect(readFileSync(countFile, 'utf8')).toBe('xx') // cache expired → new spawn - } finally { - vi.useRealTimers() - } + it('reflects an external config change on the next same-argv read', async () => { + const configFile = join(dir, 'external-config') + const countFile = join(dir, 'spawns') + writeFileSync(configFile, 'before') + fakeBin( + 'external-config.js', + `const fs = require('node:fs'); fs.appendFileSync(${JSON.stringify(countFile)}, 'x'); process.stdout.write(JSON.stringify({ value: fs.readFileSync(${JSON.stringify(configFile)}, 'utf8') }))`, + ) + + await expect(spawnCli(['model-alias', '--list'])).resolves.toEqual({ value: 'before' }) + writeFileSync(configFile, 'after') + await expect(spawnCli(['model-alias', '--list'])).resolves.toEqual({ value: 'after' }) + expect(readFileSync(countFile, 'utf8')).toBe('xx') }) it('never coalesces config-mutating action calls', async () => { @@ -427,14 +435,62 @@ describe('spawnCli coalescing (read-only)', () => { expect(readFileSync(countFile, 'utf8')).toBe('xx') // two independent spawns }) - it('flushes the read cache when an action completes, so post-action refetches are fresh', async () => { + it('runs a fresh read after a config-mutating action', async () => { const countFile = join(dir, 'spawns') fakeBin('mixed.js', `require('fs').appendFileSync(${JSON.stringify(countFile)},'x'); process.stdout.write(JSON.stringify({ok:1}))`) - await spawnCli(['model-alias', '--list']) // primes the 5s cache - await spawnCliAction(['model-alias', 'a', 'b']) // config change → cache flush - await spawnCli(['model-alias', '--list']) // must NOT serve the pre-action cache + await spawnCli(['model-alias', '--list']) + await spawnCliAction(['model-alias', 'a', 'b']) + await spawnCli(['model-alias', '--list']) expect(readFileSync(countFile, 'utf8')).toBe('xxx') }) + + it('fences old in-flight reads across a mutation without deleting the new flight', async () => { + const configFile = join(dir, 'generation-config') + const startsFile = join(dir, 'generation-read-starts') + const releaseDir = join(dir, 'generation-release') + mkdirSync(releaseDir) + writeFileSync(configFile, 'old') + fakeBin( + 'generation-fence.js', + `const fs = require('node:fs'); const path = require('node:path'); + if (process.argv[3] === '--list') { + const value = fs.readFileSync(${JSON.stringify(configFile)}, 'utf8'); + fs.appendFileSync(${JSON.stringify(startsFile)}, 'r'); + const generation = fs.readFileSync(${JSON.stringify(startsFile)}, 'utf8').length; + const release = path.join(${JSON.stringify(releaseDir)}, String(generation)); + const timer = setInterval(() => { + if (!fs.existsSync(release)) return; + clearInterval(timer); + process.stdout.write(JSON.stringify({ value, generation })); + }, 5); + } else { + fs.writeFileSync(${JSON.stringify(configFile)}, 'new'); + process.stdout.write('updated'); + }`, + ) + + const oldRead = spawnCli(['model-alias', '--list']) + await waitFor(() => readMaybe(startsFile) === 'r') + await expect(spawnCliAction(['model-alias', 'alias', 'model'])) + .resolves.toMatchObject({ ok: true }) + + const newRead = spawnCli(['model-alias', '--list']) + await waitFor(() => readMaybe(startsFile) === 'rr') + writeFileSync(join(releaseDir, '1'), '') + await expect(oldRead).resolves.toEqual({ value: 'old', generation: 1 }) + + // Settling the superseded flight must not remove the current generation's + // entry: this identical call still shares read #2 instead of spawning #3. + const coalescedNewRead = spawnCli(['model-alias', '--list']) + await new Promise(resolve => setTimeout(resolve, 100)) + expect(readMaybe(startsFile)).toBe('rr') + + writeFileSync(join(releaseDir, '2'), '') + await expect(Promise.all([newRead, coalescedNewRead])).resolves.toEqual([ + { value: 'new', generation: 2 }, + { value: 'new', generation: 2 }, + ]) + }) }) describe('resident serve single-flight', () => { @@ -561,6 +617,63 @@ describe('resident serve single-flight', () => { expect(chunks.join('')).toBe('CODEBURN_PROGRESS {"kind":"provider","provider":"claude","state":"start","generation":1}\n') }) + it('rejects and terminates a resident that emits an oversized valid JSON frame', async () => { + const startsFile = join(dir, 'oversized-frame-starts') + const oneShotsFile = join(dir, 'oversized-frame-one-shots') + fakeBin( + 'oversized-frame-resident.js', + `const fs = require('node:fs'); const readline = require('node:readline'); + if (process.argv[2] === 'serve') { + fs.appendFileSync(${JSON.stringify(startsFile)}, 's'); + const generation = fs.readFileSync(${JSON.stringify(startsFile)}, 'utf8').length; + const rl = readline.createInterface({ input: process.stdin }); + rl.once('line', line => { + const request = JSON.parse(line); + const output = generation === 1 + ? JSON.stringify({ value: 'x'.repeat(16 * 1024 * 1024 + 1024) }) + : JSON.stringify({ generation }); + process.stdout.write(JSON.stringify({ id: request.id, ok: true, output }) + '\\n'); + }); + setInterval(() => {}, 1000); + } else { + fs.appendFileSync(${JSON.stringify(oneShotsFile)}, 'o'); + process.stdout.write('{}'); + }`, + ) + startServe() + + await expect(spawnCli(['status', '--oversized-frame'], { timeoutMs: 5_000 })) + .rejects.toMatchObject({ kind: 'too-large' } satisfies Partial) + await expect(spawnCli(['status', '--after-oversized-frame'], { timeoutMs: 5_000 })) + .resolves.toEqual({ generation: 2 }) + expect(readMaybe(startsFile)).toBe('ss') + expect(readMaybe(oneShotsFile)).toBe('') + }) + + it('rejects and terminates a resident whose protocol line never terminates', async () => { + const startsFile = join(dir, 'unterminated-line-starts') + const oneShotsFile = join(dir, 'unterminated-line-one-shots') + fakeBin( + 'unterminated-line-resident.js', + `const fs = require('node:fs'); const readline = require('node:readline'); + if (process.argv[2] === 'serve') { + fs.appendFileSync(${JSON.stringify(startsFile)}, 's'); + const rl = readline.createInterface({ input: process.stdin }); + rl.once('line', () => process.stdout.write('x'.repeat(16 * 1024 * 1024 + 1024))); + setInterval(() => {}, 1000); + } else { + fs.appendFileSync(${JSON.stringify(oneShotsFile)}, 'o'); + process.stdout.write('{}'); + }`, + ) + startServe() + + await expect(spawnCli(['status', '--unterminated-line'], { timeoutMs: 5_000 })) + .rejects.toMatchObject({ kind: 'too-large' } satisfies Partial) + expect(readMaybe(startsFile)).toBe('s') + expect(readMaybe(oneShotsFile)).toBe('') + }) + it('keeps requests with any non-progress env override on the one-shot path', async () => { const files = fakeResidentBin() startServe() @@ -706,8 +819,8 @@ describe('resident serve single-flight', () => { const before = await spawnCli(['status'], { timeoutMs: 5_000 }) as { generation: number } const action = await spawnCliAction(['export', '-f', 'json', '-o', join(dir, 'usage.json')], { timeoutMs: 5_000 }) - // Different argv bypasses the 5s result cache and proves which resident - // generation actually handled the next served read. + // A different panel query proves which resident generation handled the + // next served read without relying on same-request coalescing. const after = await spawnCli(['models', '--format', 'json'], { timeoutMs: 5_000 }) as { generation: number } expect(action.ok).toBe(true) @@ -744,6 +857,26 @@ describe('killAll', () => { .resolves.toMatchObject({ ok: false, code: null }) expect(readMaybe(startsFile)).toBe('') }) + + it('terminal shutdown cancels read and action slots admitted before their spawn microtask', async () => { + const startsFile = join(dir, 'starts-after-admission') + fakeBin( + 'shutdown-after-admission.js', + `require('node:fs').appendFileSync(${JSON.stringify(startsFile)}, process.argv[2] + '\\n'); if (process.argv[2] === 'status') process.stdout.write('{}'); else process.stdout.write('updated')`, + ) + + // Both calls synchronously acquire the two free scheduler slots. Their + // actual spawn resumes in a microtask, which is exactly the before-quit race. + const read = spawnCli(['status', '--admitted']) + const action = spawnCliAction(['currency', 'EUR']) + shutdownAll() + + await Promise.all([ + expect(read).rejects.toMatchObject({ kind: 'nonzero' }), + expect(action).resolves.toMatchObject({ ok: false, code: null }), + ]) + expect(readMaybe(startsFile)).toBe('') + }) }) describe('spawnCli concurrency scheduler', () => { @@ -871,6 +1004,52 @@ describe('spawnCli concurrency scheduler', () => { await delay(50) expect(startedList(startedFile)).not.toContain('sessions') // never spawned }) + + it('limits six simultaneous resident-failure fallbacks to two one-shot children', async () => { + const startedFile = join(dir, 'fallback-started') + const activeDir = join(dir, 'fallback-active'); mkdirSync(activeDir) + const activeCountsFile = join(dir, 'fallback-active-counts') + const releaseDir = join(dir, 'fallback-release'); mkdirSync(releaseDir) + fakeBin( + 'failing-resident-with-blocked-fallbacks.js', + `const fs = require('node:fs'); const path = require('node:path'); const readline = require('node:readline'); + if (process.argv[2] === 'serve') { + const rl = readline.createInterface({ input: process.stdin }); + rl.on('line', line => { + const request = JSON.parse(line); + process.stdout.write(JSON.stringify({ id: request.id, ok: false, error: 'resident failed' }) + '\\n'); + }); + } else { + const id = process.argv[3]; + fs.appendFileSync(${JSON.stringify(startedFile)}, id + '\\n'); + const activeFile = path.join(${JSON.stringify(activeDir)}, String(process.pid)); + fs.writeFileSync(activeFile, ''); + fs.appendFileSync(${JSON.stringify(activeCountsFile)}, fs.readdirSync(${JSON.stringify(activeDir)}).length + '\\n'); + const releaseFile = path.join(${JSON.stringify(releaseDir)}, id); + const timer = setInterval(() => { + if (!fs.existsSync(releaseFile)) return; + clearInterval(timer); + fs.unlinkSync(activeFile); + process.stdout.write(JSON.stringify({ via: 'spawn', id })); + }, 5); + }`, + ) + startServe() + + const requests = Array.from({ length: 6 }, (_, index) => + spawnCli(['status', `fallback-${index}`], { timeoutMs: 5_000 }), + ) + await waitUntil(() => startedList(startedFile).length >= 2) + await delay(150) + const admittedBeforeRelease = startedList(startedFile) + + for (let index = 0; index < 6; index += 1) release(releaseDir, `fallback-${index}`) + await Promise.all(requests) + + const activeCounts = startedList(activeCountsFile).map(Number) + expect(admittedBeforeRelease).toHaveLength(2) + expect(Math.max(...activeCounts)).toBeLessThanOrEqual(2) + }) }) describe('spawnCliAction', () => { diff --git a/app/electron/cli.ts b/app/electron/cli.ts index 938f3796..6b6789f3 100644 --- a/app/electron/cli.ts +++ b/app/electron/cli.ts @@ -60,9 +60,6 @@ const DEFAULT_TIMEOUT_MS = 45_000 export const DESKTOP_COLD_TIMEOUT_MS = 10 * 60_000 // A runaway CLI (or a compromised binary) must not exhaust main-process memory. const MAX_OUTPUT_BYTES = 16 * 1024 * 1024 -// Same-cadence pollers fire near-identical read spawns; share one child and hold -// its result briefly so six overview hooks don't launch six processes at once. -const COALESCE_TTL_MS = 5_000 // A cold-cache CLI spawn costs seconds at ~120% CPU; letting every poll + // prefetch launch at once saturates the machine. Cap how many children run // concurrently — the rest queue and drain as slots free (interactive first). @@ -71,7 +68,10 @@ const MAX_CONCURRENT_CLI = 2 // Every live child so `before-quit` can reap them (Electron does not on macOS). const activeChildren = new Set() const readInflight = new Map>() -const readCache = new Map() +// Successful mutations advance the epoch before their promise resolves. A read +// begun against older config may still settle for its original caller, but can +// never be reused by the post-mutation refetch or delete that newer flight. +let readGeneration = 0 // Concurrency scheduler. `running` counts spawned (not queued) children; waiters // hold the slot-grant resolver for a queued spawn. Two queues so interactive @@ -395,6 +395,25 @@ function runCli(spec: SpawnSpec, cmdLabel: string, timeoutMs: number, onStderr?: }) } +/** Run a one-shot read under the global child cap. A slot grant resumes on a + * microtask, so terminal shutdown must be checked again immediately before the + * synchronous spawn call. */ +async function runScheduledCli( + spec: SpawnSpec, + cmdLabel: string, + timeoutMs: number, + priority: SpawnPriority, + onStderr?: (chunk: string) => void, +): Promise { + await acquireSlot(priority) + try { + if (shuttingDown) throw new CliError('nonzero', 'codeburn is shutting down') + return await runCli(spec, cmdLabel, timeoutMs, onStderr) + } finally { + releaseSlot() + } +} + /** * Spawn `codeburn ` with plain argv (never a shell), collect stdout, and * decode it as JSON. Rejects with a structured {@link CliError}: @@ -404,8 +423,9 @@ function runCli(spec: SpawnSpec, cmdLabel: string, timeoutMs: number, onStderr?: * timeout the process was killed after `timeoutMs` * too-large stdout+stderr exceeded {@link MAX_OUTPUT_BYTES} * - * Read-only, so concurrent identical calls share one child and a 5s result cache - * absorbs same-cadence pollers. Never use this for config-mutating commands. + * Read-only, so concurrent identical calls share one child. Settled results are + * never cached here because config can also change outside the desktop app. + * Never use this for config-mutating commands. */ // ── Resident serve child ──────────────────────────────────────────────── // The heavy read queries (one per panel) each pay seconds of CLI startup on @@ -430,11 +450,13 @@ class ServeClient { reject: (e: Error) => void timer: NodeJS.Timeout warmsServe: boolean + decodedBytes: number onStderr?: (chunk: string) => void }>() private nextId = 1 private deaths = 0 private buffer = '' + private bufferBytes = 0 private warmed = false private destroyed = false private requestTail: Promise = Promise.resolve() @@ -453,19 +475,27 @@ class ServeClient { child.stdout!.on('data', (chunk: string) => { // A replaced child's stream can drain after its exit callback. Never let // those stale bytes repopulate the shared line buffer for the new child. - if (this.child === child) this.onData(chunk) + if (this.child === child) this.onData(child, chunk) }) const onGone = () => this.onDeath(child) child.on('exit', onGone) child.on('error', onGone) } - private onData(chunk: string): void { + private onData(child: ReturnType, chunk: string): void { this.buffer += chunk + this.bufferBytes += Buffer.byteLength(chunk) let idx: number while ((idx = this.buffer.indexOf('\n')) >= 0) { - const line = this.buffer.slice(0, idx).trim() + const rawLine = this.buffer.slice(0, idx) this.buffer = this.buffer.slice(idx + 1) + const rawLineBytes = Buffer.byteLength(rawLine) + this.bufferBytes = Math.max(0, this.bufferBytes - rawLineBytes - 1) + if (rawLineBytes > MAX_OUTPUT_BYTES) { + this.terminateForOverflow(child) + return + } + const line = rawLine.trim() if (!line) continue let msg: { id?: number; ready?: boolean; progress?: string; ok?: boolean; refused?: boolean; output?: string; error?: string } try { msg = JSON.parse(line) } catch { continue } @@ -474,11 +504,14 @@ class ServeClient { const waiter = this.pending.get(msg.id) if (!waiter) continue if (typeof msg.progress === 'string') { + if (!this.consumeDecodedOutput(child, waiter, msg.progress)) return if (waiter.onStderr) { try { waiter.onStderr(msg.progress) } catch { /* progress consumers never own the request */ } } continue } + const terminalOutput = typeof msg.output === 'string' ? msg.output : typeof msg.error === 'string' ? msg.error : '' + if (!this.consumeDecodedOutput(child, waiter, terminalOutput)) return this.pending.delete(msg.id) clearTimeout(waiter.timer) if (msg.ok && typeof msg.output === 'string') { @@ -489,6 +522,39 @@ class ServeClient { waiter.reject(new CliError('nonzero', msg.error ?? 'serve request failed')) } } + // Complete lines are bounded above before parsing. Bound the partial frame + // too, otherwise a child that never emits '\n' can grow this buffer forever. + if (this.bufferBytes > MAX_OUTPUT_BYTES) this.terminateForOverflow(child) + } + + private consumeDecodedOutput( + child: ReturnType, + waiter: { decodedBytes: number }, + output: string, + ): boolean { + waiter.decodedBytes += Buffer.byteLength(output) + if (waiter.decodedBytes <= MAX_OUTPUT_BYTES) return true + this.terminateForOverflow(child) + return false + } + + private terminateForOverflow(child: ReturnType): void { + if (this.child !== child) return + const error = new CliError('too-large', `codeburn serve produced more than ${MAX_OUTPUT_BYTES} bytes`) + // Detach synchronously before SIGKILL. A new request may start the next + // generation immediately; the old child's eventual exit must not reject it. + this.child = null + this.buffer = '' + this.bufferBytes = 0 + this.warmed = false + this.deaths += 1 + activeChildren.delete(child as never) + for (const [, waiter] of this.pending) { + clearTimeout(waiter.timer) + waiter.reject(error) + } + this.pending.clear() + child.kill('SIGKILL') } private onDeath(child: ReturnType, countsTowardBudget = true): void { @@ -498,6 +564,7 @@ class ServeClient { if (this.child !== child) return this.child = null this.buffer = '' + this.bufferBytes = 0 this.warmed = false if (countsTowardBudget) this.deaths += 1 activeChildren.delete(child as never) @@ -547,6 +614,7 @@ class ServeClient { reject, timer, warmsServe: args[0] === 'status', + decodedBytes: 0, ...(onStderr ? { onStderr } : {}), }) child.stdin!.write(JSON.stringify({ id, args }) + '\n', (err) => { @@ -620,15 +688,16 @@ export function spawnCli( const spec = spawnSpecFor(target, args) if (opts.extraEnv) spec.env = { ...spec.env, ...opts.extraEnv } - const key = JSON.stringify([spec.bin, ...spec.args]) - const cached = readCache.get(key) - if (cached && Date.now() - cached.at < COALESCE_TTL_MS) return Promise.resolve(cached.value) + const generation = readGeneration + const key = JSON.stringify([generation, spec.bin, ...spec.args]) const existing = readInflight.get(key) // A same-cadence re-poll during a slow cold warmup coalesces onto the one // in-flight child (which already carries onStderr); no second cold parse. - // Coalesce/cache hits settle here, BEFORE queueing, so they never hold a slot. + // Coalesced calls settle here, BEFORE queueing, so they never hold a slot. if (existing) return existing + const priority = opts.priority ?? 'interactive' + // Serve fast-path: the child is started once at app startup. It accepts the // first real query before its ready frame, making that request the single // cache warm-up. CODEBURN_PROGRESS is compatible because startServe sets it @@ -644,26 +713,28 @@ export function spawnCli( .catch(err => { // App shutdown is terminal: never turn rejected resident requests // into brand-new one-shot children after killAll() has reaped them. - if (serve.isDestroyed()) throw err - return runCli(spec, args[0] ?? '', opts.timeoutMs ?? DEFAULT_TIMEOUT_MS, opts.onStderr) + if (serve.isDestroyed() || (err instanceof CliError && err.kind === 'too-large')) throw err + return runScheduledCli( + spec, + args[0] ?? '', + opts.timeoutMs ?? DEFAULT_TIMEOUT_MS, + priority, + opts.onStderr, + ) }) - .then(value => { readCache.set(key, { at: Date.now(), value }); return value }) .finally(() => { readInflight.delete(key) }) readInflight.set(key, flight) return flight } } - const priority = opts.priority ?? 'interactive' - const flight = (async () => { - await acquireSlot(priority) - try { - return await runCli(spec, args[0] ?? '', opts.timeoutMs ?? DEFAULT_TIMEOUT_MS, opts.onStderr) - } finally { - releaseSlot() - } - })() - .then(value => { readCache.set(key, { at: Date.now(), value }); return value }) + const flight = runScheduledCli( + spec, + args[0] ?? '', + opts.timeoutMs ?? DEFAULT_TIMEOUT_MS, + priority, + opts.onStderr, + ) .finally(() => { readInflight.delete(key) }) readInflight.set(key, flight) return flight @@ -686,6 +757,7 @@ export function spawnCliAction(args: string[], opts: { timeoutMs?: number } = {} return { ok: false, stdout: '', stderr: 'codeburn cancelled', code: null } } try { + if (shuttingDown) return { ok: false, stdout: '', stderr: 'codeburn is shutting down', code: null } return await runAction(spec, args, timeoutMs) } finally { releaseSlot() @@ -706,10 +778,13 @@ function runAction(spec: SpawnSpec, args: string[], timeoutMs: number): Promise< settled = true clearTimeout(timer) activeChildren.delete(child) - // The action may have changed config the read cache still reflects; a - // Settings refetch fires immediately after, so serve it fresh data. - readCache.clear() - if (result.ok && actionInvalidatesServe(args)) restartServeAfterMutation() + if (result.ok && actionInvalidatesServe(args)) { + // Fence coalescing before the action promise resolves. An immediate + // same-argv refetch belongs to the new config generation even while an + // older read is still running. + readGeneration += 1 + restartServeAfterMutation() + } resolve(result) } diff --git a/mac/Sources/CodeBurnMenubar/Data/DataClient.swift b/mac/Sources/CodeBurnMenubar/Data/DataClient.swift index 0a2c24ad..73159eed 100644 --- a/mac/Sources/CodeBurnMenubar/Data/DataClient.swift +++ b/mac/Sources/CodeBurnMenubar/Data/DataClient.swift @@ -122,14 +122,43 @@ struct DataClient { private static func runCLI( subcommand: [String], qualityOfService: QualityOfService = .userInitiated + ) async throws -> ProcessResult { + try await runCLI( + subcommand: subcommand, + serveRequest: { args in + try await ServeConnection.shared.request(args: args) + }, + spawnFallback: { + await spawnLimiter.acquire() + defer { Task { await spawnLimiter.release() } } + let process = CodeburnCLI.makeProcess( + subcommand: subcommand, + qualityOfService: qualityOfService + ) + return try await runProcess( + process, + timeoutSeconds: spawnTimeoutSeconds, + label: subcommand.joined(separator: " ") + ) + } + ) + } + + /// Internal seam for behavior-shaped lifecycle tests. Production supplies + /// the shared resident and globally limited one-shot closures above. + static func runCLI( + subcommand: [String], + serveRequest: ([String]) async throws -> Data, + spawnFallback: () async throws -> ProcessResult ) async throws -> ProcessResult { // Serve path: the first real status payload warms the resident child, // then later payloads reuse it (no node boot or session-cache reload). - // Any serve failure falls back to the spawn path below, so this remains - // strictly an optimization and takes no spawn slot. + // Transport/protocol failures fall back to the spawn path below, so + // the resident remains an optimization. Resource-policy failures stay + // terminal and cannot bypass the resident output ceiling. if ServeConnection.isEligible(subcommand) { do { - let stdout = try await ServeConnection.shared.request(args: subcommand) + let stdout = try await serveRequest(subcommand) return ProcessResult(stdout: stdout, stderr: "", exitCode: 0) } catch let error as CancellationError { // Cancellation is control flow from the refresh owner. Starting @@ -137,18 +166,25 @@ struct DataClient { // expensive cold parse and delay task teardown. throw error } catch { + if let terminalError = terminalServeError(error) { + throw terminalError + } // Resident serve is only an optimization. Protocol, child, and // timeout failures retain the established one-shot fallback, // unless a sibling teardown raced this task's cancellation. try Task.checkCancellation() } } - await spawnLimiter.acquire() - defer { Task { await spawnLimiter.release() } } - let process = CodeburnCLI.makeProcess(subcommand: subcommand, qualityOfService: qualityOfService) - return try await runProcess(process, - timeoutSeconds: spawnTimeoutSeconds, - label: subcommand.joined(separator: " ")) + return try await spawnFallback() + } + + /// Some resident failures are terminal resource-policy decisions, not + /// transport failures. Retrying those through the one-shot path would redo + /// the cold scan and could bypass the resident's stricter output ceiling. + static func terminalServeError(_ error: Error) -> DataClientError? { + guard let failure = error as? ServeConnection.ServeRequestFailed, + failure.reason == .outputTooLarge else { return nil } + return .outputTooLarge } /// Runs an already-configured process to completion, draining its output and diff --git a/mac/Sources/CodeBurnMenubar/Data/ServeConnection.swift b/mac/Sources/CodeBurnMenubar/Data/ServeConnection.swift index 25f36388..4d1f879c 100644 --- a/mac/Sources/CodeBurnMenubar/Data/ServeConnection.swift +++ b/mac/Sources/CodeBurnMenubar/Data/ServeConnection.swift @@ -10,8 +10,9 @@ import Foundation /// - The first real status request is also the warm-up. It may be written /// before the child announces READY; the pipe buffers it until serve reads /// stdin, avoiding a second one-shot process that parses the same cache. -/// - Any failure falls back to the spawn path for that call; three child -/// deaths disable serve for this app run. +/// - Transport/protocol failures fall back to the spawn path for that call; +/// resource-policy failures remain terminal. Three child deaths disable +/// serve for this app run. /// - The child's stdin closing (app quit, even SIGKILL) ends the server loop /// on the CLI side, so no orphan survives the menubar. actor ServeConnection { @@ -20,31 +21,76 @@ actor ServeConnection { typealias ProcessFactory = ([String], QualityOfService) -> Process typealias TimeoutSleep = @Sendable (UInt64) async throws -> Void + private struct QueuedRequest { + let token: Int + let args: [String] + let continuation: CheckedContinuation + } + + private struct ActiveRequest { + let token: Int + let id: Int + let args: [String] + let child: Process + } + private var process: Process? private var stdinHandle: FileHandle? private var nextId = 1 + private var nextRequestToken = 1 + private var queuedRequests: [QueuedRequest] = [] + private var activeRequest: ActiveRequest? private var pending: [Int: CheckedContinuation] = [:] + private var requestTimeouts: [Int: Task] = [:] + private var timeoutOwners: [Int: Process] = [:] + private var responseBytes: [Int: Int] = [:] private var deaths = 0 private var buffer = Data() private var receivedTerminalResponse = false + private var outputTasks: [ObjectIdentifier: Task] = [:] + private var terminationTasks: [ObjectIdentifier: Task] = [:] private let makeProcess: ProcessFactory private let timeoutSleep: TimeoutSleep + private let terminationGraceSleep: TimeoutSleep + private let responseLimitBytes: Int private static let maxDeaths = 3 + static let maxResponseBytes = 16 * 1024 * 1024 + private static let stdoutReadChunkBytes = 64 * 1024 + private static let terminationGraceNanoseconds: UInt64 = 1_000_000_000 private static let coldRequestTimeoutNanoseconds: UInt64 = 10 * 60 * 1_000_000_000 private static let warmRequestTimeoutNanoseconds: UInt64 = 60 * 1_000_000_000 struct ServeUnavailable: Error {} - struct ServeRequestFailed: Error { let message: String } + enum FailureReason: Sendable, Equatable { + case generic + case outputTooLarge + } + struct ServeRequestFailed: Error, Sendable { + let message: String + let reason: FailureReason + + init(message: String, reason: FailureReason = .generic) { + self.message = message + self.reason = reason + } + } init( makeProcess: @escaping ProcessFactory = CodeburnCLI.makeProcess, timeoutSleep: @escaping TimeoutSleep = { nanoseconds in try await Task.sleep(nanoseconds: nanoseconds) - } + }, + terminationGraceSleep: @escaping TimeoutSleep = { nanoseconds in + try await Task.sleep(nanoseconds: nanoseconds) + }, + responseLimitBytes: Int = ServeConnection.maxResponseBytes ) { self.makeProcess = makeProcess self.timeoutSleep = timeoutSleep + self.terminationGraceSleep = terminationGraceSleep + precondition(responseLimitBytes > 0) + self.responseLimitBytes = responseLimitBytes } static func isEligible(_ subcommand: [String]) -> Bool { @@ -69,18 +115,10 @@ actor ServeConnection { return } let stdoutPipe = Pipe() + let stdoutReader = stdoutPipe.fileHandleForReading child.standardInput = stdinPipe child.standardOutput = stdoutPipe child.standardError = FileHandle.nullDevice - stdoutPipe.fileHandleForReading.readabilityHandler = { [weak self] handle in - let data = handle.availableData - guard !data.isEmpty else { return } - Task { await self?.consume(data, from: child) } - } - child.terminationHandler = { [weak self] terminatedChild in - stdoutPipe.fileHandleForReading.readabilityHandler = nil - Task { await self?.childDied(terminatedChild) } - } do { try child.run() } catch { @@ -89,6 +127,28 @@ actor ServeConnection { } process = child stdinHandle = stdinWriter + let generation = ObjectIdentifier(child) + // One blocking reader owns this generation's stdout. It never reads a + // second bounded chunk until the actor has consumed the first, giving + // the 16 MiB protocol limit real backpressure instead of accumulating + // an unbounded callback/AsyncStream backlog. EOF is observed only after + // the pipe's final bytes, so child death cannot overtake a split reply. + outputTasks[generation] = Task.detached { [weak self] in + var bytes = [UInt8](repeating: 0, count: Self.stdoutReadChunkBytes) + while !Task.isCancelled { + let count = Darwin.read(stdoutReader.fileDescriptor, &bytes, bytes.count) + if count > 0 { + guard let self else { break } + await self.consume(Data(bytes[0.. Data { - guard let stdinHandle, let child = process else { throw ServeUnavailable() } + private func startNextRequestIfPossible() { + guard activeRequest == nil, !queuedRequests.isEmpty else { return } + ensureStarted() + guard let stdinHandle, let child = process else { + failQueuedRequests(error: ServeUnavailable()) + return + } + // A Process can report not-running just before its termination callback + // reaches the ordered event stream. Keep the request queued for that + // event instead of writing to a generation which is already exiting. + guard child.isRunning else { return } + + let request = queuedRequests.removeFirst() let id = nextId nextId += 1 - let request: [String: Any] = ["id": id, "args": args] - let line = try JSONSerialization.data(withJSONObject: request) - // Every request admitted before the first terminal response is a cold - // request, including concurrent startup fetches. Once any terminal - // frame arrives the resident child is hydrated and later requests use - // the ordinary one-minute guard. + let line: Data + do { + line = try JSONSerialization.data(withJSONObject: ["id": id, "args": request.args]) + } catch { + request.continuation.resume(throwing: error) + startNextRequestIfPossible() + return + } + + // The previous response can resume its caller just before EOF reaches + // this actor. Avoid admitting a successor to an already-reaped child; + // the reader's ordered EOF path will start it on a replacement. + guard child.isRunning else { + queuedRequests.insert(request, at: 0) + outputStreamEnded(for: child) + return + } + + // Select and arm the timeout only when this request becomes the sole + // protocol request in flight. A queued request must not spend its own + // budget while its predecessor is still hydrating or draining. let timeoutNanoseconds = receivedTerminalResponse ? Self.warmRequestTimeoutNanoseconds : Self.coldRequestTimeoutNanoseconds - let sleep = timeoutSleep - return try await withThrowingTaskGroup(of: Data.self) { group in - group.addTask { - try await self.registerAndWrite( - id: id, - line: line, - stdinHandle: stdinHandle, - child: child - ) - } - group.addTask { - try await sleep(timeoutNanoseconds) - // A hung request would block the serialized queue behind it: - // kill the child so everything falls back to spawns. - await self.cancelPendingRequest( - id: id, - child: child, - error: ServeRequestFailed(message: "serve timeout"), - countsAsDeath: true - ) - throw ServeRequestFailed(message: "serve timeout") - } - let result = try await group.next()! - group.cancelAll() - return result + activeRequest = ActiveRequest( + token: request.token, + id: id, + args: request.args, + child: child + ) + pending[id] = request.continuation + responseBytes[id] = 0 + do { + try stdinHandle.write(contentsOf: line + Data("\n".utf8)) + armTimeout(id: id, child: child, nanoseconds: timeoutNanoseconds) + } catch { + // The previous terminal frame can resume its caller just before + // EOF detaches that generation. Preserve this never-admitted + // request and retry it on the replacement instead of surfacing a + // transient EPIPE to the UI. + pending.removeValue(forKey: id) + responseBytes.removeValue(forKey: id) + activeRequest = nil + queuedRequests.insert(request, at: 0) + outputStreamEnded(for: child) } } - private func registerAndWrite( - id: Int, - line: Data, - stdinHandle: FileHandle, - child: Process - ) async throws -> Data { - try Task.checkCancellation() - return try await withTaskCancellationHandler { - let response = try await withCheckedThrowingContinuation { continuation in - // Register synchronously on the actor before writing. A tiny fake - // server (and occasionally a hot real child) can answer faster - // than a separately scheduled registration Task would run. - pending[id] = continuation - do { - try stdinHandle.write(contentsOf: line + Data("\n".utf8)) - } catch { - pending.removeValue(forKey: id) - continuation.resume(throwing: ServeRequestFailed(message: "stdin write failed")) - } - } - try Task.checkCancellation() - return response - } onCancel: { - Task { - await self.cancelPendingRequest( - id: id, - child: child, - error: CancellationError(), - countsAsDeath: false - ) - } + private func cancelRequest(token: Int) { + if let index = queuedRequests.firstIndex(where: { $0.token == token }) { + let request = queuedRequests.remove(at: index) + request.continuation.resume(throwing: CancellationError()) + return } - } - - private func cancelPendingRequest( - id: Int, - child: Process, - error: Error, - countsAsDeath: Bool - ) { - guard let continuation = pending.removeValue(forKey: id) else { return } - continuation.resume(throwing: error) + guard let activeRequest, activeRequest.token == token, + let continuation = pending.removeValue(forKey: activeRequest.id) else { return } + continuation.resume(throwing: CancellationError()) // Caller cancellation abandons only this response. The serialized serve // child may still be doing the expensive first hydration, and killing it // here lets tab switches and UI watchdogs restart that work indefinitely. - // A real request timeout still kills the exact child that owns the hung - // request; its termination callback consumes the death budget normally. - guard countsAsDeath, process === child, child.isRunning else { return } + // Its independent request timeout remains armed: a command that never + // returns is still reaped, so it cannot wedge every later serialized call. + } + + private func armTimeout(id: Int, child: Process, nanoseconds: UInt64) { + let sleep = timeoutSleep + timeoutOwners[id] = child + requestTimeouts[id] = Task.detached { [weak self] in + do { + try await sleep(nanoseconds) + } catch { + return + } + await self?.requestTimedOut(id: id) + } + } + + private func requestTimedOut(id: Int) { + guard let child = timeoutOwners.removeValue(forKey: id) else { return } + requestTimeouts.removeValue(forKey: id) + responseBytes.removeValue(forKey: id) + if let continuation = pending.removeValue(forKey: id) { + continuation.resume(throwing: ServeRequestFailed(message: "serve timeout")) + } + // The waiter may already have been abandoned by caller cancellation. + // Timeout ownership is deliberately independent of that continuation: + // kill only the exact generation that received the timed-out request. + guard process === child else { + if activeRequest?.id == id { activeRequest = nil } + startNextRequestIfPossible() + return + } + // Retire the timed-out generation synchronously. Its stdout may never + // reach EOF (for example, a stuck child can ignore SIGTERM or a + // descendant can retain the pipe), so waiting for the reader would also + // spend every queued caller's timeout before it can even be admitted. + process = nil + stdinHandle = nil + buffer = Data() + receivedTerminalResponse = false + deaths += 1 + if activeRequest?.id == id { activeRequest = nil } + cancelTimeouts(ownedBy: child) + terminateTimedOutChild(child) + // The waiter was removed above and cannot be requeued by stale EOF. + // A queued read starts on a replacement immediately, subject to the + // ordinary three-death budget. + startNextRequestIfPossible() + } + + private func cancelTimeout(id: Int) { + timeoutOwners.removeValue(forKey: id) + requestTimeouts.removeValue(forKey: id)?.cancel() + responseBytes.removeValue(forKey: id) + } + + private func cancelTimeouts(ownedBy child: Process) { + let ids = timeoutOwners.compactMap { id, owner in owner === child ? id : nil } + for id in ids { cancelTimeout(id: id) } + } + + private func cancelAllTimeouts() { + for task in requestTimeouts.values { task.cancel() } + requestTimeouts.removeAll() + timeoutOwners.removeAll() + responseBytes.removeAll() + } + + private func outputStreamFinished(for child: Process) { + outputTasks.removeValue(forKey: ObjectIdentifier(child)) + if !child.isRunning { + terminationTasks.removeValue(forKey: ObjectIdentifier(child))?.cancel() + } + } + + private func terminateTimedOutChild(_ child: Process) { + guard child.isRunning else { return } child.terminate() + let generation = ObjectIdentifier(child) + let sleep = terminationGraceSleep + terminationTasks[generation] = Task.detached { [weak self] in + do { + try await sleep(Self.terminationGraceNanoseconds) + } catch { + // Cancellation means the owner stopped waiting: either shutdown + // (which must not orphan a SIGTERM-ignoring generation) or the + // child already died and the stream finished. Escalate either + // way; the isRunning guard makes the dead-child case a no-op. + await self?.forceKillAfterGrace(child) + return + } + await self?.forceKillAfterGrace(child) + } + } + + private func forceKillAfterGrace(_ child: Process) { + terminationTasks.removeValue(forKey: ObjectIdentifier(child)) + guard child.isRunning else { return } + _ = Darwin.kill(child.processIdentifier, SIGKILL) + } + + private func outputStreamEnded(for child: Process) { + guard process === child else { return } + // EOF/read failure is a transport death even if the process has not + // reaped yet. Terminate that exact generation so a child which closed + // stdout cannot survive after the actor starts its replacement. + if child.isRunning { child.terminate() } + childDied(child) } // Internal so the generation guard can be exercised deterministically by @@ -213,34 +383,101 @@ actor ServeConnection { // old process exits. If a replacement starts first, those late bytes must // not repopulate the shared line buffer or mark the new child as warm. guard process === child else { return } - buffer.append(data) - while let newline = buffer.firstIndex(of: UInt8(ascii: "\n")) { - let lineData = buffer.subdata(in: buffer.startIndex.. Bool { + guard let current = responseBytes[id], + count <= responseLimitBytes - current else { + outputOverflowed(child) + return false } + responseBytes[id] = current + count + return true + } + + private func outputOverflowed(_ child: Process) { + guard process === child else { return } + // Detach this exact generation before terminating it. Its eventual exit + // and any already-scheduled stdout callbacks are then stale and cannot + // consume a second death or corrupt a replacement generation. + process = nil + stdinHandle = nil + buffer = Data() + receivedTerminalResponse = false + deaths += 1 + cancelTimeouts(ownedBy: child) + failAllRequests(error: ServeRequestFailed( + message: "serve output exceeded \(responseLimitBytes) bytes", + reason: .outputTooLarge + )) + if child.isRunning { child.terminate() } } private func childDied(_ child: Process) { @@ -250,13 +487,43 @@ actor ServeConnection { buffer.removeAll() receivedTerminalResponse = false deaths += 1 - failAllPending() + cancelTimeouts(ownedBy: child) + if let activeRequest, activeRequest.child === child { + if let continuation = pending.removeValue(forKey: activeRequest.id) { + // Only read-only status requests enter this connection. If a + // generation exits after admission but before its terminal + // reply, retain the waiter and retry on the replacement rather + // than racing it into a one-shot fallback. A timed-out or + // cancelled waiter is already absent and is never retried. + queuedRequests.insert(QueuedRequest( + token: activeRequest.token, + args: activeRequest.args, + continuation: continuation + ), at: 0) + } + self.activeRequest = nil + } + // Requests which were never written survive an ordinary child crash. + // They begin on a replacement only after this ordered death event. + startNextRequestIfPossible() } - private func failAllPending() { + private func failAllRequests( + error: Error = ServeRequestFailed(message: "serve exited") + ) { for (_, continuation) in pending { - continuation.resume(throwing: ServeRequestFailed(message: "serve exited")) + continuation.resume(throwing: error) } pending.removeAll() + activeRequest = nil + failQueuedRequests(error: error) + } + + private func failQueuedRequests(error: Error) { + let requests = queuedRequests + queuedRequests.removeAll() + for request in requests { + request.continuation.resume(throwing: error) + } } } diff --git a/mac/Tests/CodeBurnMenubarTests/ServeConnectionTests.swift b/mac/Tests/CodeBurnMenubarTests/ServeConnectionTests.swift index ca74ca8c..a5c7588e 100644 --- a/mac/Tests/CodeBurnMenubarTests/ServeConnectionTests.swift +++ b/mac/Tests/CodeBurnMenubarTests/ServeConnectionTests.swift @@ -6,6 +6,7 @@ import Testing private let ignoredSIGPIPEHandlerBits = unsafeBitCast(SIG_IGN, to: UInt.self) private let coldTimeoutNanoseconds: UInt64 = 10 * 60 * 1_000_000_000 private let warmTimeoutNanoseconds: UInt64 = 60 * 1_000_000_000 +private let terminationGraceNanoseconds: UInt64 = 1_000_000_000 private func currentSIGPIPEHandlerBits() -> UInt { var action = sigaction() @@ -18,8 +19,8 @@ private actor TimeoutRecorder { func recordAndSleep(_ nanoseconds: UInt64) async throws { values.append(nanoseconds) - // Cold timers stay pending until the fake child replies and the task - // group cancels them. The warm timer returns immediately to exercise + // Cold timers stay pending until the fake child replies and the + // connection cancels them. The warm timer returns immediately to exercise // the timeout path without a real one-minute wait. if nanoseconds == warmTimeoutNanoseconds { return } try await Task.sleep(nanoseconds: 5 * 1_000_000_000) @@ -35,6 +36,59 @@ private actor TimeoutRecorder { func snapshot() -> [UInt64] { values } } +private actor FallbackRecorder { + private var calls = 0 + + func record() { calls += 1 } + func snapshot() -> Int { calls } +} + +/// A cancellation-aware timeout clock that tests can advance explicitly. This +/// keeps the regression independent of the production ten-minute cold budget. +private actor ManualTimeoutClock { + private struct Waiter { + let nanoseconds: UInt64 + let continuation: CheckedContinuation + } + + private var nextToken = 0 + private var waiters: [Int: Waiter] = [:] + private var recorded: [UInt64] = [] + + func sleep(_ nanoseconds: UInt64) async throws { + let token = nextToken + nextToken += 1 + recorded.append(nanoseconds) + try await withTaskCancellationHandler { + try await withCheckedThrowingContinuation { (continuation: CheckedContinuation) in + if Task.isCancelled { + continuation.resume(throwing: CancellationError()) + } else { + waiters[token] = Waiter(nanoseconds: nanoseconds, continuation: continuation) + } + } + } onCancel: { + Task { await self.cancel(token) } + } + } + + func snapshot() -> [UInt64] { + waiters.keys.sorted().compactMap { waiters[$0]?.nanoseconds } + } + + func history() -> [UInt64] { recorded } + + func fireOldest() { + guard let token = waiters.keys.min(), let waiter = waiters.removeValue(forKey: token) else { return } + waiter.continuation.resume() + } + + private func cancel(_ token: Int) { + guard let waiter = waiters.removeValue(forKey: token) else { return } + waiter.continuation.resume(throwing: CancellationError()) + } +} + private final class QualityOfServiceRecorder: @unchecked Sendable { private let lock = NSLock() private var values: [QualityOfService] = [] @@ -52,6 +106,29 @@ private final class QualityOfServiceRecorder: @unchecked Sendable { } } +private final class ProcessQueue: @unchecked Sendable { + private let lock = NSLock() + private var processes: [Process] + + init(_ processes: [Process]) { + self.processes = processes + } + + func take(qualityOfService: QualityOfService) -> Process { + lock.lock() + let child = processes.removeFirst() + lock.unlock() + child.qualityOfService = qualityOfService + return child + } + + var remainingCount: Int { + lock.lock() + defer { lock.unlock() } + return processes.count + } +} + @Suite("ServeConnection", .serialized) struct ServeConnectionTests { @Test("the resident child starts at user-initiated QoS") @@ -141,7 +218,7 @@ struct ServeConnectionTests { return child }, timeoutSleep: { nanoseconds in - try await recorder.recordAndSleep(nanoseconds) + try await recorder.recordAndWait(nanoseconds) } ) @@ -164,22 +241,20 @@ struct ServeConnectionTests { } // Submit the next request while the child is still blocked hydrating - // the cancelled first one. Two timeout selections prove both requests - // reached send() before the fake is released to emit either response. + // the cancelled first one. It stays client-side queued: neither its + // stdin line nor its own timeout may begin yet. let second = Task { try await connection.request(args: ["status", "--request", "second"]) } - for _ in 0..<200 { - if await recorder.snapshot().count >= 2 { break } - try await Task.sleep(nanoseconds: 10_000_000) - } - #expect(await recorder.snapshot().count == 2) + try await Task.sleep(nanoseconds: 100_000_000) + #expect(await recorder.snapshot() == [coldTimeoutNanoseconds]) #expect(try String(contentsOfFile: eventsFile, encoding: .utf8) == "first-read\n") _ = FileManager.default.createFile(atPath: releaseMarker, contents: Data()) let secondPayload = try await second.value #expect(String(decoding: secondPayload, as: UTF8.self) == "live-2") + #expect(await recorder.snapshot() == [coldTimeoutNanoseconds, warmTimeoutNanoseconds]) let pids = try String(contentsOfFile: pidsFile, encoding: .utf8) .split(separator: "\n") #expect(pids.count == 1) @@ -189,6 +264,256 @@ struct ServeConnectionTests { await connection.shutdown() } + @Test("a cancelled never-returning request retains a timeout owner and cannot wedge later work") + func cancelledHungRequestIsEventuallyReaped() async throws { + let dir = NSTemporaryDirectory() + "serve-connection-cancel-timeout-test-" + UUID().uuidString + try FileManager.default.createDirectory(atPath: dir, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(atPath: dir) } + let firstReadMarker = dir + "/first-read" + let pidsFile = dir + "/pids" + let clock = ManualTimeoutClock() + let graceClock = ManualTimeoutClock() + + let stuckChild = Process() + stuckChild.executableURL = URL(fileURLWithPath: "/bin/sh") + stuckChild.arguments = ["-c", """ + trap '' TERM + printf '%s\n' "$$" >> "$1" + IFS= read -r line + : > "$2" + while :; do :; done + """, "serve-fixture", pidsFile, firstReadMarker] + + let replacement = Process() + replacement.executableURL = URL(fileURLWithPath: "/bin/sh") + replacement.arguments = ["-c", """ + printf '%s\n' "$$" >> "$1" + while IFS= read -r line; do + id=$(printf '%s' "$line" | sed -E 's/.*"id":([0-9]+).*/\\1/') + printf '{"id":%s,"ok":true,"output":"replacement-%s"}\n' "$id" "$id" + done + """, "serve-fixture", pidsFile] + + let children = ProcessQueue([stuckChild, replacement]) + let connection = ServeConnection( + makeProcess: { _, qualityOfService in + children.take(qualityOfService: qualityOfService) + }, + timeoutSleep: { nanoseconds in + try await clock.sleep(nanoseconds) + }, + terminationGraceSleep: { nanoseconds in + try await graceClock.sleep(nanoseconds) + } + ) + defer { + if stuckChild.isRunning { _ = Darwin.kill(stuckChild.processIdentifier, SIGKILL) } + } + + let abandoned = Task { + try await connection.request(args: ["status", "--request", "stuck"]) + } + for _ in 0..<200 where !FileManager.default.fileExists(atPath: firstReadMarker) { + try await Task.sleep(nanoseconds: 10_000_000) + } + #expect(FileManager.default.fileExists(atPath: firstReadMarker)) + #expect(await clock.snapshot() == [coldTimeoutNanoseconds]) + + abandoned.cancel() + do { + _ = try await abandoned.value + #expect(Bool(false), "cancelled request unexpectedly succeeded") + } catch { + #expect(error is CancellationError) + } + + // The caller is gone, but the independently-owned cold timeout must + // remain armed. This assertion is the red-before regression: the old + // task-group race cancelled the only timeout along with the caller. + #expect(await clock.snapshot() == [coldTimeoutNanoseconds]) + let successor = Task { + try await connection.request(args: ["status", "--request", "after-cancel"]) + } + try await Task.sleep(nanoseconds: 100_000_000) + #expect(await clock.snapshot() == [coldTimeoutNanoseconds]) + #expect(children.remainingCount == 1) + + await clock.fireOldest() + for _ in 0..<200 where children.remainingCount > 0 { + try await Task.sleep(nanoseconds: 10_000_000) + } + let replacementStartedBeforeOldEOF = children.remainingCount == 0 + #expect(replacementStartedBeforeOldEOF) + // Keep the red-before run finite: the old implementation waits for EOF + // forever because this fixture deliberately ignores SIGTERM. + if !replacementStartedBeforeOldEOF { + _ = Darwin.kill(stuckChild.processIdentifier, SIGKILL) + for _ in 0..<200 where children.remainingCount > 0 { + try await Task.sleep(nanoseconds: 10_000_000) + } + } + + // The retired child ignores SIGTERM, yet its stale stdout remains open. + // The queued successor must already run on a replacement; it cannot wait + // for either old-generation EOF or the force-kill grace period. + let payload = try await successor.value + #expect(String(decoding: payload, as: UTF8.self) == "replacement-2") + #expect(await clock.snapshot().isEmpty) + #expect(await clock.history() == [coldTimeoutNanoseconds, coldTimeoutNanoseconds]) + #expect(await graceClock.snapshot() == [terminationGraceNanoseconds]) + #expect(stuckChild.isRunning) + #expect(try String(contentsOfFile: pidsFile, encoding: .utf8).split(separator: "\n").count == 2) + + await graceClock.fireOldest() + for _ in 0..<200 where stuckChild.isRunning { + try await Task.sleep(nanoseconds: 10_000_000) + } + #expect(!stuckChild.isRunning) + #expect(stuckChild.terminationReason == .uncaughtSignal) + #expect(stuckChild.terminationStatus == SIGKILL) + await connection.shutdown() + } + + @Test("shutdown during the termination grace force-kills the SIGTERM-ignoring generation") + func shutdownDuringGraceKillsStubbornChild() async throws { + let dir = NSTemporaryDirectory() + "serve-connection-shutdown-grace-test-" + UUID().uuidString + try FileManager.default.createDirectory(atPath: dir, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(atPath: dir) } + let firstReadMarker = dir + "/first-read" + let clock = ManualTimeoutClock() + let graceClock = ManualTimeoutClock() + + let stuckChild = Process() + stuckChild.executableURL = URL(fileURLWithPath: "/bin/sh") + stuckChild.arguments = ["-c", """ + trap '' TERM + IFS= read -r line + : > "$1" + while :; do :; done + """, "serve-fixture", firstReadMarker] + + let children = ProcessQueue([stuckChild]) + let connection = ServeConnection( + makeProcess: { _, qualityOfService in + children.take(qualityOfService: qualityOfService) + }, + timeoutSleep: { nanoseconds in + try await clock.sleep(nanoseconds) + }, + terminationGraceSleep: { nanoseconds in + try await graceClock.sleep(nanoseconds) + } + ) + defer { + if stuckChild.isRunning { _ = Darwin.kill(stuckChild.processIdentifier, SIGKILL) } + } + + let request = Task { + try await connection.request(args: ["status", "--request", "stuck"]) + } + for _ in 0..<200 where !FileManager.default.fileExists(atPath: firstReadMarker) { + try await Task.sleep(nanoseconds: 10_000_000) + } + #expect(FileManager.default.fileExists(atPath: firstReadMarker)) + #expect(await clock.snapshot() == [coldTimeoutNanoseconds]) + + // Time out the request: the generation is retired and SIGTERM'd, and the + // SIGKILL escalation parks on the injected grace clock. + await clock.fireOldest() + do { + _ = try await request.value + #expect(Bool(false), "timed-out request unexpectedly succeeded") + } catch let error as ServeConnection.ServeRequestFailed { + #expect(error.message == "serve timeout") + } + for _ in 0..<200 where await graceClock.snapshot().isEmpty { + try await Task.sleep(nanoseconds: 10_000_000) + } + #expect(await graceClock.snapshot() == [terminationGraceNanoseconds]) + #expect(stuckChild.isRunning) // SIGTERM ignored; escalation still pending + + // Shutdown must not merely cancel the escalation. The retired generation + // is already detached from `process`, so nothing else will reap it; the + // grace task's cancellation path has to SIGKILL it or it outlives the app. + await connection.shutdown() + for _ in 0..<200 where stuckChild.isRunning { + try await Task.sleep(nanoseconds: 10_000_000) + } + #expect(!stuckChild.isRunning) + #expect(stuckChild.terminationReason == .uncaughtSignal) + #expect(stuckChild.terminationStatus == SIGKILL) + } + + @Test("timed-out generations consume one death each and stop at the resident budget") + func timeoutDeathBudgetIsExact() async throws { + let dir = NSTemporaryDirectory() + "serve-connection-timeout-budget-test-" + UUID().uuidString + try FileManager.default.createDirectory(atPath: dir, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(atPath: dir) } + let readsFile = dir + "/reads" + let clock = ManualTimeoutClock() + let processes = (0..<3).map { _ in + let child = Process() + child.executableURL = URL(fileURLWithPath: "/bin/sh") + child.arguments = ["-c", """ + trap '' TERM + IFS= read -r line + printf r >> "$1" + while :; do :; done + """, "serve-fixture", readsFile] + return child + } + let children = ProcessQueue(processes) + let connection = ServeConnection( + makeProcess: { _, qualityOfService in + children.take(qualityOfService: qualityOfService) + }, + timeoutSleep: { nanoseconds in + try await clock.sleep(nanoseconds) + }, + terminationGraceSleep: { _ in } + ) + defer { + for child in processes where child.isRunning { + _ = Darwin.kill(child.processIdentifier, SIGKILL) + } + } + + for attempt in 0..<3 { + let request = Task { + try await connection.request(args: ["status", "--attempt", String(attempt)]) + } + for _ in 0..<200 { + let reads = (try? String(contentsOfFile: readsFile, encoding: .utf8).count) ?? 0 + if reads == attempt + 1, await clock.snapshot().count == 1 { break } + try await Task.sleep(nanoseconds: 10_000_000) + } + #expect((try? String(contentsOfFile: readsFile, encoding: .utf8).count) == attempt + 1) + await clock.fireOldest() + do { + _ = try await request.value + Issue.record("timeout \(attempt) unexpectedly succeeded") + } catch let error as ServeConnection.ServeRequestFailed { + #expect(error.message == "serve timeout") + } + } + + #expect(children.remainingCount == 0) + do { + _ = try await connection.request(args: ["status", "--after-budget"]) + Issue.record("resident restarted after three timed-out generations") + } catch { + #expect(error is ServeConnection.ServeUnavailable) + } + for _ in 0..<200 where processes.contains(where: \.isRunning) { + try await Task.sleep(nanoseconds: 10_000_000) + } + #expect(processes.allSatisfy { !$0.isRunning }) + #expect(processes.allSatisfy { + $0.terminationReason == .uncaughtSignal && $0.terminationStatus == SIGKILL + }) + await connection.shutdown() + } + @Test("external cancellations keep one child and safely discard late replies") func cancellationsKeepResidentChildAlive() async throws { let dir = NSTemporaryDirectory() + "serve-connection-cancel-reuse-test-" + UUID().uuidString @@ -258,6 +583,105 @@ struct ServeConnectionTests { await connection.shutdown() } + @Test("cancelling a queued request never writes it or arms its timeout") + func queuedCancellationNeverReachesChild() async throws { + let dir = NSTemporaryDirectory() + "serve-connection-queued-cancel-test-" + UUID().uuidString + try FileManager.default.createDirectory(atPath: dir, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(atPath: dir) } + let requestsFile = dir + "/requests" + let releaseMarker = dir + "/release" + let recorder = TimeoutRecorder() + + let connection = ServeConnection( + makeProcess: { _, qualityOfService in + let child = Process() + child.executableURL = URL(fileURLWithPath: "/bin/sh") + child.arguments = ["-c", """ + count=0 + while IFS= read -r line; do + count=$((count + 1)) + printf '%s\n' "$line" >> "$1" + if [ "$count" -eq 1 ]; then + while [ ! -f "$2" ]; do sleep 0.01; done + fi + id=$(printf '%s' "$line" | sed -E 's/.*"id":([0-9]+).*/\\1/') + printf '{"id":%s,"ok":true,"output":"served-%s"}\n' "$id" "$id" + done + """, "serve-fixture", requestsFile, releaseMarker] + child.qualityOfService = qualityOfService + return child + }, + timeoutSleep: { nanoseconds in + try await recorder.recordAndWait(nanoseconds) + } + ) + + let first = Task { try await connection.request(args: ["status", "first"]) } + for _ in 0..<200 where !(FileManager.default.fileExists(atPath: requestsFile)) { + try await Task.sleep(nanoseconds: 10_000_000) + } + let cancelled = Task { try await connection.request(args: ["status", "cancelled"]) } + let third = Task { try await connection.request(args: ["status", "third"]) } + try await Task.sleep(nanoseconds: 100_000_000) + cancelled.cancel() + do { + _ = try await cancelled.value + Issue.record("queued cancellation unexpectedly succeeded") + } catch { + #expect(error is CancellationError) + } + #expect(await recorder.snapshot() == [coldTimeoutNanoseconds]) + + _ = FileManager.default.createFile(atPath: releaseMarker, contents: Data()) + #expect(String(decoding: try await first.value, as: UTF8.self) == "served-1") + #expect(String(decoding: try await third.value, as: UTF8.self) == "served-2") + let requests = try String(contentsOfFile: requestsFile, encoding: .utf8) + #expect(requests.contains("first")) + #expect(requests.contains("third")) + #expect(!requests.contains("cancelled")) + #expect(await recorder.snapshot() == [coldTimeoutNanoseconds, warmTimeoutNanoseconds]) + await connection.shutdown() + } + + @Test("shutdown fails the active request and every client-side queued request") + func shutdownDrainsClientQueue() async throws { + let dir = NSTemporaryDirectory() + "serve-connection-shutdown-queue-test-" + UUID().uuidString + try FileManager.default.createDirectory(atPath: dir, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(atPath: dir) } + let requestMarker = dir + "/request-read" + let recorder = TimeoutRecorder() + let connection = ServeConnection( + makeProcess: { _, qualityOfService in + let child = Process() + child.executableURL = URL(fileURLWithPath: "/bin/sh") + child.arguments = ["-c", "IFS= read -r line; : > \"$1\"; sleep 5", "serve-fixture", requestMarker] + child.qualityOfService = qualityOfService + return child + }, + timeoutSleep: { nanoseconds in + try await recorder.recordAndWait(nanoseconds) + } + ) + + let active = Task { try await connection.request(args: ["status", "active"]) } + for _ in 0..<200 where !FileManager.default.fileExists(atPath: requestMarker) { + try await Task.sleep(nanoseconds: 10_000_000) + } + let queued = Task { try await connection.request(args: ["status", "queued"]) } + try await Task.sleep(nanoseconds: 100_000_000) + #expect(await recorder.snapshot() == [coldTimeoutNanoseconds]) + await connection.shutdown() + + for request in [active, queued] { + do { + _ = try await request.value + Issue.record("shutdown request unexpectedly succeeded") + } catch { + #expect(error is ServeConnection.ServeRequestFailed) + } + } + } + @Test("late stdout from a replaced child cannot corrupt or warm its replacement") func staleGenerationStdoutIsDiscarded() async throws { let oldChild = Process() @@ -273,27 +697,22 @@ struct ServeConnectionTests { done """] - var children = [oldChild, newChild] + let children = ProcessQueue([oldChild, newChild]) let recorder = TimeoutRecorder() let connection = ServeConnection( makeProcess: { _, qualityOfService in - let child = children.removeFirst() - child.qualityOfService = qualityOfService - return child + children.take(qualityOfService: qualityOfService) }, timeoutSleep: { nanoseconds in - try await recorder.recordAndSleep(nanoseconds) + try await recorder.recordAndWait(nanoseconds) } ) - do { - _ = try await connection.request(args: ["status", "--generation", "old"]) - #expect(Bool(false), "old child unexpectedly answered") - } catch { - #expect(error is ServeConnection.ServeRequestFailed) - } - - await connection.ensureStarted() + // The admitted read survives the old generation's crash and retries + // on the replacement. Request id 1 belonged to the old child; the + // replacement receives id 2. + let retried = try await connection.request(args: ["status", "--generation", "old"]) + #expect(String(decoding: retried, as: UTF8.self) == "new-2") // Model both harmful trailing shapes after the replacement owns the // connection: a complete terminal would incorrectly select the warm @@ -306,13 +725,17 @@ struct ServeConnectionTests { let payload = try await connection.request(args: ["status", "--generation", "new"]) - #expect(String(decoding: payload, as: UTF8.self) == "new-2") - #expect(await recorder.snapshot() == [coldTimeoutNanoseconds, coldTimeoutNanoseconds]) - #expect(children.isEmpty) + #expect(String(decoding: payload, as: UTF8.self) == "new-3") + #expect(await recorder.snapshot() == [ + coldTimeoutNanoseconds, + coldTimeoutNanoseconds, + warmTimeoutNanoseconds, + ]) + #expect(children.remainingCount == 0) await connection.shutdown() } - @Test("all concurrent cold requests get ten minutes, then warm requests get one minute") + @Test("queued requests arm their warm timeout only after cold hydration finishes") func coldAndWarmTimeoutSelection() async throws { let dir = NSTemporaryDirectory() + "serve-connection-timeout-test-" + UUID().uuidString try FileManager.default.createDirectory(atPath: dir, withIntermediateDirectories: true) @@ -326,32 +749,25 @@ struct ServeConnectionTests { child.executableURL = URL(fileURLWithPath: "/bin/sh") child.arguments = ["-c", """ IFS= read -r first - IFS= read -r second while [ ! -f "$1" ]; do sleep 0.01; done - for line in "$first" "$second"; do + for slot in first second third; do + if [ "$slot" = first ]; then line="$first"; else IFS= read -r line; fi id=$(printf '%s' "$line" | sed -E 's/.*"id":([0-9]+).*/\\1/') printf '{"id":%s,"ok":true,"output":"served"}\\n' "$id" done - IFS= read -r third - sleep 2 """, "serve-fixture", releaseMarker] child.qualityOfService = qualityOfService return child }, timeoutSleep: { nanoseconds in - try await recorder.recordAndSleep(nanoseconds) + try await recorder.recordAndWait(nanoseconds) } ) let first = Task { try await connection.request(args: ["status", "--request", "one"]) } let second = Task { try await connection.request(args: ["status", "--request", "two"]) } - for _ in 0..<200 { - if await recorder.snapshot().count >= 2 { break } - try await Task.sleep(nanoseconds: 10_000_000) - } - let coldSelections = await recorder.snapshot() - #expect(coldSelections.count == 2) - #expect(coldSelections.allSatisfy { $0 == coldTimeoutNanoseconds }) + try await Task.sleep(nanoseconds: 100_000_000) + #expect(await recorder.snapshot() == [coldTimeoutNanoseconds]) _ = FileManager.default.createFile(atPath: releaseMarker, contents: Data()) let firstPayload = try await first.value @@ -359,16 +775,12 @@ struct ServeConnectionTests { #expect(String(decoding: firstPayload, as: UTF8.self) == "served") #expect(String(decoding: secondPayload, as: UTF8.self) == "served") - do { - _ = try await connection.request(args: ["status", "--request", "three"]) - #expect(Bool(false), "warm request unexpectedly escaped its timeout") - } catch { - #expect(error is ServeConnection.ServeRequestFailed) - } + let thirdPayload = try await connection.request(args: ["status", "--request", "three"]) + #expect(String(decoding: thirdPayload, as: UTF8.self) == "served") let allSelections = await recorder.snapshot() #expect(allSelections == [ coldTimeoutNanoseconds, - coldTimeoutNanoseconds, + warmTimeoutNanoseconds, warmTimeoutNanoseconds, ]) await connection.shutdown() @@ -421,6 +833,230 @@ struct ServeConnectionTests { await connection.shutdown() } + @Test("an actual stdout flood is bounded and the next generation stays healthy") + func oversizedFrameTerminatesOnlyItsGeneration() async throws { + let oldChild = Process() + oldChild.executableURL = URL(fileURLWithPath: "/bin/sh") + oldChild.arguments = ["-c", """ + IFS= read -r line + dd if=/dev/zero bs=1024 count=1 2>/dev/null | tr '\\0' x + sleep 5 + """] + + let replacement = Process() + replacement.executableURL = URL(fileURLWithPath: "/bin/sh") + replacement.arguments = ["-c", """ + IFS= read -r line + id=$(printf '%s' "$line" | sed -E 's/.*"id":([0-9]+).*/\\1/') + printf '{"id":%s,"ok":true,"output":"replacement"}\\n' "$id" + """] + + let children = ProcessQueue([oldChild, replacement]) + let recorder = TimeoutRecorder() + let connection = ServeConnection( + makeProcess: { _, qualityOfService in + children.take(qualityOfService: qualityOfService) + }, + timeoutSleep: { nanoseconds in + try await recorder.recordAndWait(nanoseconds) + }, + responseLimitBytes: 128 + ) + + do { + _ = try await connection.request(args: ["status", "--oversized"]) + #expect(Bool(false), "oversized resident frame unexpectedly succeeded") + } catch let error as ServeConnection.ServeRequestFailed { + #expect(error.reason == .outputTooLarge) + } + + await connection.ensureStarted() + let payload = try await connection.request(args: ["status", "--replacement"]) + #expect(String(decoding: payload, as: UTF8.self) == "replacement") + #expect(children.remainingCount == 0) + await connection.shutdown() + } + + @Test("an unterminated frame and cumulative progress cannot bypass the resident limit") + func partialAndCumulativeFramesAreBounded() async throws { + for mode in ["partial", "progress"] { + let child = Process() + child.executableURL = URL(fileURLWithPath: "/bin/sh") + child.arguments = ["-c", "IFS= read -r line; sleep 5"] + let recorder = TimeoutRecorder() + let connection = ServeConnection( + makeProcess: { _, qualityOfService in + child.qualityOfService = qualityOfService + return child + }, + timeoutSleep: { nanoseconds in + try await recorder.recordAndWait(nanoseconds) + }, + responseLimitBytes: 128 + ) + let request = Task { try await connection.request(args: ["status", "--mode", mode]) } + for _ in 0..<200 { + if await recorder.snapshot().count == 1 { break } + try await Task.sleep(nanoseconds: 10_000_000) + } + + if mode == "partial" { + await connection.consume(Data(repeating: UInt8(ascii: "x"), count: 129), from: child) + } else { + let progress = String(repeating: "p", count: 70) + let frame = Data("{\"id\":1,\"progress\":\"\(progress)\"}\n".utf8) + #expect(frame.count < 128) + await connection.consume(frame, from: child) + await connection.consume(frame, from: child) + } + + do { + _ = try await request.value + #expect(Bool(false), "\(mode) overflow unexpectedly succeeded") + } catch let error as ServeConnection.ServeRequestFailed { + #expect(error.reason == .outputTooLarge) + } + await connection.shutdown() + } + } + + @Test("a cancelled request keeps its cumulative progress bound until the child finishes") + func cancelledRequestStillBoundsOrphanProgress() async throws { + let oldChild = Process() + oldChild.executableURL = URL(fileURLWithPath: "/bin/sh") + oldChild.arguments = ["-c", "IFS= read -r line; sleep 5"] + + let replacement = Process() + replacement.executableURL = URL(fileURLWithPath: "/bin/sh") + replacement.arguments = ["-c", """ + IFS= read -r line + id=$(printf '%s' "$line" | sed -E 's/.*"id":([0-9]+).*/\\1/') + printf '{"id":%s,"ok":true,"output":"healthy"}\\n' "$id" + """] + + let children = ProcessQueue([oldChild, replacement]) + let recorder = TimeoutRecorder() + let connection = ServeConnection( + makeProcess: { _, qualityOfService in + children.take(qualityOfService: qualityOfService) + }, + timeoutSleep: { nanoseconds in + try await recorder.recordAndWait(nanoseconds) + }, + responseLimitBytes: 128 + ) + + let abandoned = Task { try await connection.request(args: ["status", "--abandoned"]) } + for _ in 0..<200 { + if await recorder.snapshot().count == 1 { break } + try await Task.sleep(nanoseconds: 10_000_000) + } + abandoned.cancel() + do { + _ = try await abandoned.value + Issue.record("cancelled request unexpectedly succeeded") + } catch { + #expect(error is CancellationError) + } + + let progress = String(repeating: "p", count: 70) + let frame = Data("{\"id\":1,\"progress\":\"\(progress)\"}\n".utf8) + await connection.consume(frame, from: oldChild) + await connection.consume(frame, from: oldChild) + + await connection.ensureStarted() + #expect(children.remainingCount == 0) + let payload = try await connection.request(args: ["status", "--replacement"]) + #expect(String(decoding: payload, as: UTF8.self) == "healthy") + await connection.shutdown() + } + + @Test("each overflow consumes exactly one resident death") + func overflowDeathBudgetIsExact() async throws { + let processes = (0..<3).map { _ in + let child = Process() + child.executableURL = URL(fileURLWithPath: "/bin/sh") + child.arguments = ["-c", "IFS= read -r line; sleep 5"] + return child + } + let children = ProcessQueue(processes) + let recorder = TimeoutRecorder() + let connection = ServeConnection( + makeProcess: { _, qualityOfService in + children.take(qualityOfService: qualityOfService) + }, + timeoutSleep: { nanoseconds in + try await recorder.recordAndWait(nanoseconds) + }, + responseLimitBytes: 64 + ) + + for attempt in 0..<3 { + let request = Task { try await connection.request(args: ["status", "--attempt", "\(attempt)"]) } + for _ in 0..<200 { + if await recorder.snapshot().count == attempt + 1 { break } + try await Task.sleep(nanoseconds: 10_000_000) + } + await connection.consume(Data(repeating: UInt8(ascii: "x"), count: 65), from: processes[attempt]) + do { + _ = try await request.value + Issue.record("overflow \(attempt) unexpectedly succeeded") + } catch let error as ServeConnection.ServeRequestFailed { + #expect(error.reason == .outputTooLarge) + } + } + + #expect(children.remainingCount == 0) + do { + _ = try await connection.request(args: ["status", "--after-budget"]) + Issue.record("resident restarted after exhausting its death budget") + } catch { + #expect(error is ServeConnection.ServeUnavailable) + } + await connection.shutdown() + } + + @Test("output overflow is not eligible for a one-shot fallback") + func outputOverflowIsTerminalForDataClient() async { + let overflow = ServeConnection.ServeRequestFailed( + message: "too large", + reason: .outputTooLarge + ) + let fallback = FallbackRecorder() + do { + _ = try await DataClient.runCLI( + subcommand: ["status", "--format", "menubar-json"], + serveRequest: { _ in throw overflow }, + spawnFallback: { + await fallback.record() + return DataClient.ProcessResult(stdout: Data(), stderr: "", exitCode: 0) + } + ) + Issue.record("output overflow unexpectedly fell back or succeeded") + } catch DataClientError.outputTooLarge { + // Expected: the one-shot closure must remain untouched. + } catch { + Issue.record("unexpected terminal error: \(error)") + } + #expect(await fallback.snapshot() == 0) + + let ordinary = ServeConnection.ServeRequestFailed(message: "serve exited") + do { + let result = try await DataClient.runCLI( + subcommand: ["status", "--format", "menubar-json"], + serveRequest: { _ in throw ordinary }, + spawnFallback: { + await fallback.record() + return DataClient.ProcessResult(stdout: Data("fallback".utf8), stderr: "", exitCode: 0) + } + ) + #expect(String(decoding: result.stdout, as: UTF8.self) == "fallback") + } catch { + Issue.record("ordinary serve failure did not use fallback: \(error)") + } + #expect(await fallback.snapshot() == 1) + } + @Test("the first real request is the only cold-start query") func firstRequestIsTheWarmup() async throws { let dir = NSTemporaryDirectory() + "serve-connection-test-" + UUID().uuidString @@ -456,6 +1092,39 @@ struct ServeConnectionTests { await connection.shutdown() } + @Test("split terminal bytes are drained before child death and the next generation stays clean") + func finalStdoutDrainPrecedesTermination() async throws { + let first = Process() + first.executableURL = URL(fileURLWithPath: "/bin/sh") + first.arguments = ["-c", """ + IFS= read -r line + id=$(printf '%s' "$line" | sed -E 's/.*"id":([0-9]+).*/\\1/') + printf '{"id":%s,"ok":true,' "$id" + printf '"output":"final-drain"}\n' + """] + + let replacement = Process() + replacement.executableURL = URL(fileURLWithPath: "/bin/sh") + replacement.arguments = ["-c", """ + while IFS= read -r line; do + id=$(printf '%s' "$line" | sed -E 's/.*"id":([0-9]+).*/\\1/') + printf '{"id":%s,"ok":true,"output":"replacement"}\n' "$id" + done + """] + + let children = ProcessQueue([first, replacement]) + let connection = ServeConnection { _, qualityOfService in + children.take(qualityOfService: qualityOfService) + } + + let drained = try await connection.request(args: ["status", "drain"]) + #expect(String(decoding: drained, as: UTF8.self) == "final-drain") + let next = try await connection.request(args: ["status", "next"]) + #expect(String(decoding: next, as: UTF8.self) == "replacement") + #expect(children.remainingCount == 0) + await connection.shutdown() + } + @Test("a child that closes stdin fails the request without terminating the app") func closedChildStdinDoesNotRaiseSIGPIPE() async throws { let dir = NSTemporaryDirectory() + "serve-connection-sigpipe-test-" + UUID().uuidString diff --git a/src/codex-cache.ts b/src/codex-cache.ts index 7676dc7b..5ba09ea3 100644 --- a/src/codex-cache.ts +++ b/src/codex-cache.ts @@ -1,7 +1,8 @@ import { readFile, mkdir, stat, open, rename, unlink } from 'fs/promises' import { existsSync } from 'fs' import { randomBytes } from 'crypto' -import { join } from 'path' +import { join, resolve } from 'path' +import { AsyncLocalStorage } from 'node:async_hooks' import { getCodeburnCacheDir } from './cache-dir.js' import type { ParsedProviderCall } from './providers/types.js' @@ -31,24 +32,43 @@ type ResultCache = { files: Record } -function getCachePath(): string { - return join(getCodeburnCacheDir(), CACHE_FILE) +const cacheDirContext = new AsyncLocalStorage() + +function currentCacheDir(): string { + return cacheDirContext.getStore() ?? resolve(getCodeburnCacheDir()) } -let memCache: ResultCache | null = null +// A parse can cross many async boundaries before the Codex provider publishes +// its incremental cache. Embedded hosts are allowed to change the process env +// between calls, so pin the call-time directory for the whole transaction +// instead of re-reading CODEBURN_CACHE_DIR at each cache operation. +export function withCodexCacheDirectory(cacheDir: string, operation: () => T): T { + return cacheDirContext.run(resolve(cacheDir), operation) +} -async function loadCache(): Promise { - if (memCache) return memCache +function getCachePath(cacheDir: string): string { + return join(cacheDir, CACHE_FILE) +} + +// Embedded consumers can change CODEBURN_CACHE_DIR without reloading this +// module. Keep each directory's in-memory state separate so a warm cache (or an +// unflushed update) from A can never be read from or written into B. +const memCaches = new Map() + +async function loadCache(cacheDir: string): Promise { + const inMemory = memCaches.get(cacheDir) + if (inMemory) return inMemory try { - const raw = await readFile(getCachePath(), 'utf-8') + const raw = await readFile(getCachePath(cacheDir), 'utf-8') const cache = JSON.parse(raw) as ResultCache if (cache.version === CODEX_CACHE_VERSION && cache.files && typeof cache.files === 'object') { - memCache = cache + memCaches.set(cacheDir, cache) return cache } } catch {} - memCache = { version: CODEX_CACHE_VERSION, files: {} } - return memCache + const empty = { version: CODEX_CACHE_VERSION, files: {} } + memCaches.set(cacheDir, empty) + return empty } function getEntry(cache: ResultCache, filePath: string, fp: FileFingerprint): FileEntry | null { @@ -65,7 +85,7 @@ export async function readCachedCodexResults( ): Promise { try { const s = await stat(filePath) - const cache = await loadCache() + const cache = await loadCache(currentCacheDir()) const entry = getEntry(cache, filePath, { mtimeMs: s.mtimeMs, sizeBytes: s.size }) return entry?.calls ?? null } catch {} @@ -77,7 +97,7 @@ export async function getCachedCodexProject( ): Promise { try { const s = await stat(filePath) - const cache = await loadCache() + const cache = await loadCache(currentCacheDir()) const entry = getEntry(cache, filePath, { mtimeMs: s.mtimeMs, sizeBytes: s.size }) return entry?.project ?? null } catch {} @@ -102,7 +122,7 @@ export async function writeCachedCodexResults( fingerprint: FileFingerprint, ): Promise { try { - const cache = await loadCache() + const cache = await loadCache(currentCacheDir()) cache.files[filePath] = { mtimeMs: fingerprint.mtimeMs, sizeBytes: fingerprint.sizeBytes, @@ -113,6 +133,8 @@ export async function writeCachedCodexResults( } export async function flushCodexCache(): Promise { + const cacheDir = currentCacheDir() + const memCache = memCaches.get(cacheDir) if (!memCache) return try { // Evict entries for files that no longer exist on disk @@ -125,9 +147,8 @@ export async function flushCodexCache(): Promise { } } - const dir = getCodeburnCacheDir() - if (!existsSync(dir)) await mkdir(dir, { recursive: true }) - const finalPath = getCachePath() + if (!existsSync(cacheDir)) await mkdir(cacheDir, { recursive: true }) + const finalPath = getCachePath(cacheDir) const tempPath = `${finalPath}.${randomBytes(8).toString('hex')}.tmp` const payload = JSON.stringify(memCache) const handle = await open(tempPath, 'w', 0o600) diff --git a/src/parser.ts b/src/parser.ts index 712295b0..8c9dc9c5 100644 --- a/src/parser.ts +++ b/src/parser.ts @@ -6,10 +6,11 @@ import { calculateCost, calculateLocalModelSavings, getShortModelName, isProxied import { resolveSubagentAttribution, sessionIdentity } from './sessions-report.js' import { normalizeContentBlocks } from './content-utils.js' import { discoverAllSessions, getProvider } from './providers/index.js' -import { flushCodexCache } from './codex-cache.js' +import { flushCodexCache, withCodexCacheDirectory } from './codex-cache.js' import { antigravityCascadeIdFromPath, flushAntigravityCache, shouldReparseAntigravitySource } from './providers/antigravity.js' -import { getDesktopSessionsDirs } from './providers/claude.js' +import { getClaudeConfigDirs, getDesktopSessionsDirs } from './providers/claude.js' import { isSqliteBusyError } from './sqlite.js' +import { getCodeburnCacheDir } from './cache-dir.js' import { type CachedCall, type CachedFile, @@ -2869,6 +2870,10 @@ async function parseProviderSources( ): Promise { const provider = await getProvider(providerName) if (!provider) return [] + // The environment is a call-time input. Capture Antigravity's cache target + // for this whole parse transaction so a host changing CODEBURN_CACHE_DIR + // before the final flush cannot redirect A's dirty state into (or past) B. + const antigravityCacheDir = providerName === 'antigravity' ? getCodeburnCacheDir() : undefined const section = getOrCreateProviderSection(diskCache, providerName) const allDiscoveredFiles = new Set() @@ -3019,7 +3024,7 @@ async function parseProviderSources( if (didParse && providerName === 'codex') await flushCodexCache() if (didParse && providerName === 'antigravity') { const liveIds = new Set(sources.map(s => antigravityCascadeIdFromPath(s.path))) - await flushAntigravityCache(liveIds) + await flushAntigravityCache(liveIds, antigravityCacheDir) } } @@ -3190,7 +3195,15 @@ async function parseProviderSources( const CACHE_TTL_MS = 180_000 const MAX_CACHE_ENTRIES = 10 -const sessionCache = new Map() +type SessionCacheEntry = { + data: ProjectSummary[] + createdAt: number + validatedFrom: number + startMs?: number + endMs?: number + sig?: string +} +const sessionCache = new Map() // Burst reuse for a resident process (codeburn serve). Every payload command // anchors its range end at its own `new Date()`, so two panel fetches issued @@ -3207,15 +3220,16 @@ function parseBurstWindowMs(): number { // A resident process (codeburn serve) can install a validator that answers // "has any watched session root changed since this timestamp?" — typically -// backed by fs.watch over every provider's probeRoots(). While the validator -// reports clean, a previous parse stays reusable well past the burst window, -// bounded by a hard cap so a missed filesystem event self-heals instead of -// pinning stale data forever. Null (the default everywhere but serve) keeps -// reuse strictly inside the burst window. -let parseReuseValidator: ((sinceTs: number) => boolean) | null = null +// backed by fs.watch over every provider's probeRoots(). Clean extends reuse +// to the hard cap, dirty rejects every memo, and unknown (watcher coverage is +// unavailable or began too late) falls back to the ordinary exact TTL / short +// burst rather than disabling caching. Null keeps those ordinary semantics. +export type ParseReuseValidation = 'clean' | 'dirty' | 'unknown' +type ParseReuseValidator = (sinceTs: number) => ParseReuseValidation +let parseReuseValidator: ParseReuseValidator | null = null const VALIDATED_REUSE_CAP_MS = 5 * 60 * 1000 -export function setParseReuseValidator(validator: ((sinceTs: number) => boolean) | null): void { +export function setParseReuseValidator(validator: ParseReuseValidator | null): void { parseReuseValidator = validator } @@ -3227,9 +3241,13 @@ function burstReuse(dateRange: DateRange, sig: string): ProjectSummary[] | null const endMs = dateRange.end.getTime() for (const entry of sessionCache.values()) { if (entry.sig !== sig || entry.startMs !== startMs || entry.endMs === undefined) continue - const age = now - entry.ts + const validation = parseReuseValidator?.(entry.validatedFrom) ?? 'unknown' + // A dirty event during the producing parse must not be hidden even by the + // short burst. Unknown coverage, however, retains that bounded fallback. + if (validation === 'dirty') continue + const age = now - entry.createdAt const insideBurst = age <= windowMs - const validatedClean = parseReuseValidator !== null && age <= VALIDATED_REUSE_CAP_MS && parseReuseValidator(entry.ts) + const validatedClean = validation === 'clean' && age <= VALIDATED_REUSE_CAP_MS if (!insideBurst && !validatedClean) continue if (endMs < entry.endMs || endMs - entry.endMs > Math.max(windowMs, validatedClean ? VALIDATED_REUSE_CAP_MS : 0)) continue return filterProjectsByDateRange(entry.data, dateRange) @@ -3237,34 +3255,35 @@ function burstReuse(dateRange: DateRange, sig: string): ProjectSummary[] | null return null } -function cacheKey(dateRange?: DateRange, providerFilter?: string): string { +function cacheKey(dateRange: DateRange | undefined, providerFilter: string | undefined, claudeDiscoveryRoots: readonly string[]): string { const s = dateRange ? `${dateRange.start.getTime()}:${dateRange.end.getTime()}` : 'none' - // Include the Claude config-dir env so a config change in a long-lived - // process (menubar / GNOME extension / test workers) does not return - // stale data keyed under a previous configuration. - const claudeEnv = (process.env['CLAUDE_CONFIG_DIRS'] ?? '') + '|' + (process.env['CLAUDE_CONFIG_DIR'] ?? '') + // Key on the effective roots, not only their env inputs: GUI consumers can + // change config.json claudeConfigDirs while a resident serve process stays + // alive. Normalized roots also collapse syntactically different inputs that + // discover the same directories. + const claudeRoots = JSON.stringify(claudeDiscoveryRoots) // Proxy attribution (totalProxiedCostUSD) is computed live from proxyPaths and // then cached, so the key must change when that config changes. // Pricing-affecting config participates so a memoized parse (exact-key or // burst-reused in a resident serve process) can never present costs priced // under aliases/overrides/savings the user has since changed. - return `${s}:${providerFilter ?? 'all'}:${claudeEnv}:${getProxyPathsConfigHash()}:${getModelAliasesConfigHash()}:${getPriceOverridesConfigHash()}:${getLocalModelSavingsConfigHash()}` + return `${s}:${providerFilter ?? 'all'}:${claudeRoots}:${getProxyPathsConfigHash()}:${getModelAliasesConfigHash()}:${getPriceOverridesConfigHash()}:${getLocalModelSavingsConfigHash()}` } export function clearSessionCache(): void { sessionCache.clear() } -function cachePut(key: string, data: ProjectSummary[]) { +function cachePut(key: string, data: ProjectSummary[], parseStartedAt: number) { const now = Date.now() for (const [k, v] of sessionCache) { - if (now - v.ts > CACHE_TTL_MS) sessionCache.delete(k) + if (now - v.createdAt > CACHE_TTL_MS) sessionCache.delete(k) } if (sessionCache.size >= MAX_CACHE_ENTRIES) { - const oldest = [...sessionCache.entries()].sort((a, b) => a[1].ts - b[1].ts)[0] + const oldest = [...sessionCache.entries()].sort((a, b) => a[1].createdAt - b[1].createdAt)[0] if (oldest) sessionCache.delete(oldest[0]) } - sessionCache.set(key, { data, ts: now, ...(putMeta ?? {}) }) + sessionCache.set(key, { data, createdAt: now, validatedFrom: parseStartedAt, ...(putMeta ?? {}) }) putMeta = null } @@ -3707,13 +3726,33 @@ export function isSessionHydrationComplete(): boolean { // chart (gapStart = lastComputedDate + 1 never looks back at them). let readOnlyServedStale = false -export async function parseAllSessions(dateRange?: DateRange, providerFilter?: string): Promise { - const key = cacheKey(dateRange, providerFilter) +export function parseAllSessions(dateRange?: DateRange, providerFilter?: string): Promise { + // Capture synchronously, before the first await. AsyncLocalStorage keeps all + // Codex cache reads, dirty writes, and the final flush on this call-time + // directory even if an embedding host changes the process env mid-parse. + const codexCacheDir = getCodeburnCacheDir() + return withCodexCacheDirectory(codexCacheDir, () => parseAllSessionsInCacheScope(dateRange, providerFilter)) +} + +async function parseAllSessionsInCacheScope(dateRange?: DateRange, providerFilter?: string): Promise { + // Anchor freshness before any config, cache, or session input is read. A + // watched-root event that lands while this parse is in flight must remain + // newer than the resulting memo instead of being blessed retroactively. + const parseStartedAt = Date.now() + const claudeDiscoveryRoots = await getClaudeConfigDirs() + const key = cacheKey(dateRange, providerFilter, claudeDiscoveryRoots) const cached = sessionCache.get(key) - if (cached && Date.now() - cached.ts < CACHE_TTL_MS) return cached.data + if (cached) { + const age = Date.now() - cached.createdAt + const validation = parseReuseValidator?.(cached.validatedFrom) ?? 'unknown' + if ( + validation !== 'dirty' + && (age < CACHE_TTL_MS || (validation === 'clean' && age <= VALIDATED_REUSE_CAP_MS)) + ) return cached.data + } // The signature is the key minus the range: what must match for a burst // reuse (provider, config env, proxy hash) regardless of the now-anchor. - const burstSig = cacheKey(undefined, providerFilter) + const burstSig = cacheKey(undefined, providerFilter, claudeDiscoveryRoots) if (dateRange) { const reused = burstReuse(dateRange, burstSig) if (reused) return reused @@ -3735,7 +3774,7 @@ export async function parseAllSessions(dateRange?: DateRange, providerFilter?: s if (hydration.waited) diskCache = await loadCache() const isCold = !isCacheComplete(diskCache) try { - return await runParse(key, diskCache, dateRange, providerFilter, { isCold }) + return await runParse(key, diskCache, dateRange, providerFilter, { isCold, burstSig, parseStartedAt }) } finally { await hydration.release() } @@ -3747,20 +3786,20 @@ export async function parseAllSessions(dateRange?: DateRange, providerFilter?: s const priorSnapshot = diskCache const refresh = await acquireCacheRefreshLock() if (refresh.outcome === 'timed-out' || refresh.outcome === 'unavailable') { - return runParse(key, priorSnapshot, dateRange, providerFilter, { readOnly: true }) + return runParse(key, priorSnapshot, dateRange, providerFilter, { readOnly: true, burstSig, parseStartedAt }) } if (refresh.outcome === 'completed-by-other') { - return runParse(key, await loadCache(), dateRange, providerFilter, { readOnly: true }) + return runParse(key, await loadCache(), dateRange, providerFilter, { readOnly: true, burstSig, parseStartedAt }) } try { // Reload only after ownership is canonical; this closes the lost-update // window between the pre-gate read and the holder's completed publication. diskCache = await loadCache() - return await runParse(key, diskCache, dateRange, providerFilter, { refreshLock: refresh.handle }) + return await runParse(key, diskCache, dateRange, providerFilter, { refreshLock: refresh.handle, burstSig, parseStartedAt }) } catch (err) { if (!(err instanceof RefreshFenceLostError) && !(err instanceof RefreshPublicationUnavailableError)) throw err - return runParse(key, await loadCache(), dateRange, providerFilter, { readOnly: true }) + return runParse(key, await loadCache(), dateRange, providerFilter, { readOnly: true, burstSig, parseStartedAt }) } finally { await refresh.handle.release() } @@ -3773,14 +3812,16 @@ type RunParseOptions = { isCold?: boolean readOnly?: boolean refreshLock?: RefreshLockHandle + burstSig: string + parseStartedAt: number } async function runParse( key: string, diskCache: SessionCache, - dateRange?: DateRange, - providerFilter?: string, - options: RunParseOptions = {}, + dateRange: DateRange | undefined, + providerFilter: string | undefined, + options: RunParseOptions, ): Promise { const { isCold = false, readOnly = false, refreshLock } = options readOnlyServedStale = false @@ -3942,7 +3983,7 @@ async function runParse( const result = Array.from(mergedMap.values()).sort((a, b) => b.totalCostUSD - a.totalCostUSD) correlateCrossProviderPrSessions(result) - if (dateRange) setCachePutMeta({ startMs: dateRange.start.getTime(), endMs: dateRange.end.getTime(), sig: cacheKey(undefined, providerFilter) }) - cachePut(key, result) + if (dateRange) setCachePutMeta({ startMs: dateRange.start.getTime(), endMs: dateRange.end.getTime(), sig: options.burstSig }) + cachePut(key, result, options.parseStartedAt) return result } diff --git a/src/providers/antigravity.ts b/src/providers/antigravity.ts index 49eb295d..c56d6e7b 100644 --- a/src/providers/antigravity.ts +++ b/src/providers/antigravity.ts @@ -1,7 +1,7 @@ import { readdir, readFile, mkdir, stat, open, rename, unlink } from 'fs/promises' import { execFile } from 'child_process' import { randomBytes } from 'crypto' -import { basename, join } from 'path' +import { basename, join, resolve } from 'path' import { homedir } from 'os' import { fileURLToPath } from 'url' import https from 'https' @@ -162,8 +162,8 @@ type AntigravityGenMetadataRow = { const cachedServers = new Map() const cachedModelMaps = new Map() -let memCache: AntigravityCache | null = null -let cacheDirty = false +type AntigravityCacheState = { cache: AntigravityCache; dirty: boolean } +const cacheStates = new Map() let httpsAgent: https.Agent | undefined const protoTextDecoder = new TextDecoder('utf-8', { fatal: false }) @@ -176,8 +176,12 @@ function getAgent(): https.Agent { return httpsAgent } -function getCachePath(): string { - return join(getCodeburnCacheDir(), 'antigravity-results.json') +function currentCacheDir(): string { + return resolve(getCodeburnCacheDir()) +} + +function getCachePath(cacheDir: string): string { + return join(cacheDir, 'antigravity-results.json') } export function getAntigravityStatusLineEventsPath(): string { @@ -320,22 +324,30 @@ export function extractAntigravityGeneratorMetadata(resp: unknown): GeneratorMet return Array.isArray(metadata) ? metadata : [] } -async function loadCache(): Promise { - if (memCache) return memCache +async function loadCache(cacheDir: string): Promise { + const inMemory = cacheStates.get(cacheDir) + if (inMemory) return inMemory try { - const raw = await readFile(getCachePath(), 'utf-8') + const raw = await readFile(getCachePath(cacheDir), 'utf-8') const cache = JSON.parse(raw) as AntigravityCache if (cache.version === CACHE_VERSION && cache.cascades && typeof cache.cascades === 'object') { - memCache = cache - return cache + const state = { cache, dirty: false } + cacheStates.set(cacheDir, state) + return state } } catch { /* no cache or invalid */ } - memCache = { version: CACHE_VERSION, cascades: {} } - return memCache + const state: AntigravityCacheState = { + cache: { version: CACHE_VERSION, cascades: {} }, + dirty: false, + } + cacheStates.set(cacheDir, state) + return state } -async function flushCache(liveCascadeIds?: Set): Promise { - if (!memCache) return +async function flushCache(liveCascadeIds?: Set, cacheDir = currentCacheDir()): Promise { + const state = cacheStates.get(cacheDir) + if (!state) return + const memCache = state.cache // If the caller supplied liveCascadeIds, we must run the eviction step // even when no cascade was added or updated this run; otherwise deleted // .pb files would persist in the cache forever once it stops getting @@ -345,16 +357,14 @@ async function flushCache(liveCascadeIds?: Set): Promise { for (const id of Object.keys(memCache.cascades)) { if (!liveCascadeIds.has(id)) { delete memCache.cascades[id] - cacheDirty = true + state.dirty = true } } } - if (!cacheDirty) return + if (!state.dirty) return try { - - const dir = getCodeburnCacheDir() - await mkdir(dir, { recursive: true }) - const finalPath = getCachePath() + await mkdir(cacheDir, { recursive: true }) + const finalPath = getCachePath(cacheDir) const tempPath = `${finalPath}.${randomBytes(8).toString('hex')}.tmp` const handle = await open(tempPath, 'w', 0o600) try { @@ -368,7 +378,7 @@ async function flushCache(liveCascadeIds?: Set): Promise { } catch { try { await unlink(tempPath) } catch { /* cleanup */ } } - cacheDirty = false + state.dirty = false } catch { /* best-effort */ } } @@ -1170,7 +1180,9 @@ export async function snapshotAntigravityStatusLinePayload(input: unknown): Prom const s = await stat(source.path).catch(() => null) if (!s) return false - const cache = await loadCache() + const cacheDir = currentCacheDir() + const state = await loadCache(cacheDir) + const cache = state.cache const cached = cache.cascades[cascadeId] if (cached && cached.mtimeMs === s.mtimeMs && cached.sizeBytes === s.size && cached.calls.length > 0) { return true @@ -1192,8 +1204,8 @@ export async function snapshotAntigravityStatusLinePayload(input: unknown): Prom sizeBytes: s.size, calls: snapshotCalls, } - cacheDirty = true - await flushCache() + state.dirty = true + await flushCache(undefined, cacheDir) return cache.cascades[cascadeId]!.calls.length > 0 } catch { return false @@ -1297,7 +1309,8 @@ function createParser(source: SessionSource, seenKeys: Set): SessionPars } const cascadeId = antigravityCascadeIdFromPath(source.path) - const cache = await loadCache() + const state = await loadCache(currentCacheDir()) + const cache = state.cache const s = await stat(source.path).catch(() => null) if (!s) return @@ -1328,7 +1341,7 @@ function createParser(source: SessionSource, seenKeys: Set): SessionPars sizeBytes: s.size, calls: sqliteResults, } - cacheDirty = true + state.dirty = true for (const call of sqliteResults) { if (seenKeys.has(call.deduplicationKey)) continue @@ -1381,7 +1394,7 @@ function createParser(source: SessionSource, seenKeys: Set): SessionPars sizeBytes: s.size, calls: results, } - cacheDirty = true + state.dirty = true for (const call of results) { if (seenKeys.has(call.deduplicationKey)) continue @@ -1442,8 +1455,8 @@ export function createAntigravityProvider(): Provider { } } -export async function flushAntigravityCache(liveCascadeIds?: Set): Promise { - await flushCache(liveCascadeIds) +export async function flushAntigravityCache(liveCascadeIds?: Set, cacheDir?: string): Promise { + await flushCache(liveCascadeIds, cacheDir ? resolve(cacheDir) : currentCacheDir()) } export const antigravity = createAntigravityProvider() diff --git a/src/serve.ts b/src/serve.ts index 5444464b..a5f3f778 100644 --- a/src/serve.ts +++ b/src/serve.ts @@ -5,6 +5,7 @@ import { createInterface } from 'readline' import type { Command } from 'commander' import { getConfigFilePath } from './config.js' +import type { ParseReuseValidation } from './parser.js' // --------------------------------------------------------------------------- // codeburn serve --stdio: a resident query server for the desktop app. @@ -30,16 +31,77 @@ import { getConfigFilePath } from './config.js' // already guards between processes. // --------------------------------------------------------------------------- -// First-token allowlist of the app's heavy read queries. Deliberately absent: -// every config mutation (currency, model-alias set, budget, price-override, -// proxy-path, plan), export (writes files), share/devices (network + pairing -// state), menubar/web/mcp/guard/sync/act (process management or writes). // Past this resident-set size the serve loop drops its in-memory memos and // re-parses on the next request. 3GB leaves generous room for the largest // observed corpora while bounding a pathological one. const SERVE_MAX_RSS_BYTES = 3 * 1024 * 1024 * 1024 -const SERVE_COMMANDS = new Set(['status', 'overview', 'models', 'sessions', 'compare', 'yield', 'spend', 'optimize', 'audit']) +type OutputMemoEntry = { + createdAt: number + validatedFrom: number + output: string + configFingerprint: string +} + +// Kept as a small seam so the ordering contract can be tested without relying +// on filesystem watcher scheduling: an event arriving while a parse is in +// flight must be newer than the memo produced by that parse. +export function createOutputMemoEntry( + parseStartedAt: number, + parseCompletedAt: number, + output: string, + configFingerprint: string, +): OutputMemoEntry { + return { createdAt: parseCompletedAt, validatedFrom: parseStartedAt, output, configFingerprint } +} + +type ServeOptionKind = 'flag' | 'value' + +// This is intentionally a positive, command-specific option schema rather +// than a shared denylist. If a command later gains a write-capable option it +// remains a normal one-shot CLI action until it is explicitly reviewed here. +// The entries mirror the Commander definitions in main.ts. In particular, +// optimize omits its apply-only surface (--apply, --yes, --dry-run, --only). +const SERVE_OPTIONS: Readonly>>> = { + status: { + '--format': 'value', '--scope': 'value', '--provider': 'value', '--project': 'value', + '--exclude': 'value', '--period': 'value', '--day': 'value', '--from': 'value', + '--to': 'value', '--days': 'value', '--no-optimize': 'flag', '--no-timeline': 'flag', + '--claude-config-source': 'value', + }, + overview: { + '-p': 'value', '--period': 'value', '--from': 'value', '--to': 'value', + '--provider': 'value', '--project': 'value', '--exclude': 'value', '--no-color': 'flag', + }, + models: { + '-p': 'value', '--period': 'value', '--from': 'value', '--to': 'value', + '--provider': 'value', '--task': 'value', '--by-task': 'flag', '--by-agent': 'flag', + '--top': 'value', '--min-cost': 'value', '--no-totals': 'flag', '--format': 'value', + }, + sessions: { + '-p': 'value', '--period': 'value', '--from': 'value', '--to': 'value', + '--provider': 'value', '--format': 'value', '--by-pr': 'flag', '--no-pager': 'flag', + }, + compare: { + '-p': 'value', '--period': 'value', '--provider': 'value', '--format': 'value', + '--model-a': 'value', '--model-b': 'value', + }, + yield: { + '-p': 'value', '--period': 'value', '--provider': 'value', '--format': 'value', + }, + spend: { + '-p': 'value', '--period': 'value', '--from': 'value', '--to': 'value', + '--provider': 'value', '--format': 'value', + }, + optimize: { + '-p': 'value', '--period': 'value', '--from': 'value', '--to': 'value', + '--provider': 'value', '--format': 'value', '--json': 'flag', + }, + audit: { + '-p': 'value', '--period': 'value', '--from': 'value', '--to': 'value', + '--provider': 'value', '--format': 'value', + }, +} type ServeRequest = { id: string | number; args: string[] } @@ -52,10 +114,29 @@ function isServeRequest(value: unknown): value is ServeRequest { function allowed(args: string[]): boolean { const first = args[0] - if (!first || !SERVE_COMMANDS.has(first)) return false - // No request may smuggle a second positional that turns a read into - // something else; the allowed commands take flags only. - return args.slice(1).every((a, i, all) => a.startsWith('-') || (i > 0 && all[i - 1]!.startsWith('--'))) + if (!first) return false + const options = SERVE_OPTIONS[first] + if (!options) return false + + // Served commands have no positional arguments. Long options may use the + // standard --name=value form; otherwise every value must immediately + // follow an option declared as value-bearing in that command's schema. + for (let i = 1; i < args.length; i++) { + const token = args[i]! + const separator = token.startsWith('--') ? token.indexOf('=') : -1 + const option = separator >= 0 ? token.slice(0, separator) : token + const inlineValue = separator >= 0 + const kind = options[option] + if (!kind) return false + if (kind === 'flag') { + if (inlineValue) return false + continue + } + if (inlineValue) continue + const value = args[++i] + if (value === undefined || value.startsWith('-')) return false + } + return true } class ExitSignal extends Error { @@ -137,12 +218,32 @@ async function getConfigFingerprint(): Promise { /// Watch every provider's probe roots (the same paths codeburn doctor reports /// as "where discovery looks") so the parse-reuse validator can answer "did /// any session data change since T?" without a stat sweep. macOS fs.watch -/// rides FSEvents and supports recursive directory watches; a root that fails -/// to watch is simply not covered, which only shortens reuse (the burst -/// window and the hard cap still apply), never staleness. -async function startRootWatchers(): Promise<{ startedAt: number; lastEventAt: () => number; close: () => void }> { +/// rides FSEvents and supports recursive directory watches. A probe failure or +/// a watch failure for an existing root disables event-driven reuse for this +/// generation; a root absent at setup is rechecked by the parser's hard cap. +type RootWatcherState = { + startedAt: number + lastEventAt: () => number + healthy: () => boolean + close: () => void +} + +export function classifyRootReuse( + sinceTs: number, + state: { startedAt: number; lastEventAt: number; healthy: boolean }, +): ParseReuseValidation { + // A known event is conclusive even if watcher coverage degraded afterward. + // Unknown means only that no dirty evidence exists and cleanliness cannot be + // established for the whole interval. + if (state.lastEventAt >= sinceTs) return 'dirty' + if (!state.healthy || sinceTs < state.startedAt) return 'unknown' + return 'clean' +} + +async function startRootWatchers(): Promise { let lastEventAt = 0 - const startedAt = Date.now() + let healthy = true + let closed = false const watchers: FSWatcher[] = [] try { const { getAllProviders } = await import('./providers/index.js') @@ -152,21 +253,54 @@ async function startRootWatchers(): Promise<{ startedAt: number; lastEventAt: () if (!provider.probeRoots) continue try { for (const root of await provider.probeRoots()) roots.add(root.path) - } catch { /* a failing probe just goes unwatched */ } + } catch { + // An unknown probe result could hide an existing input root, so no + // global all-roots-quiet claim is safe for this watcher generation. + healthy = false + } } for (const root of roots) { + let info: Awaited> + try { + info = await stat(root) + } catch (err) { + // An absent discovery root contains no sessions at arm time. If it is + // created later there is no child watcher to see that creation, so the + // parser's hard reuse cap remains the eventual revalidation backstop. + // Other stat failures mean an existing input could be uncovered. + if ((err as NodeJS.ErrnoException).code !== 'ENOENT') healthy = false + continue + } try { - const info = await stat(root) const watcher = watch(root, { recursive: info.isDirectory() }, () => { lastEventAt = Date.now() }) - watcher.on('error', () => { /* dropped watcher = shorter reuse, never staleness */ }) + watcher.on('error', () => { healthy = false }) watchers.push(watcher) - } catch { /* nonexistent root: nothing to watch */ } + } catch { + // stat proved this input exists, so failing to arm it invalidates the + // global quiet predicate even when other roots remain watched. + healthy = false + } } - } catch { /* watcherless serve still works via the burst window */ } + } catch { + // Discovery itself failed. Existing watchers are still closed normally, + // but they cannot validate reuse for an incomplete root set. + healthy = false + } + if (watchers.length === 0) return null + + // Coverage begins only after at least one watcher has been successfully + // armed. A parse performed while provider probing/stat/watch setup was in + // flight must not be blessed retroactively as watched. + const startedAt = Date.now() return { startedAt, lastEventAt: () => lastEventAt, - close: () => { for (const w of watchers) w.close() }, + healthy: () => healthy && !closed, + close: () => { + if (closed) return + closed = true + for (const w of watchers) w.close() + }, } } @@ -180,16 +314,32 @@ export async function runStdioServe(buildProgram: () => Command): Promise // parse stays valid past the burst window (capped in parser.ts, so a missed // filesystem event self-heals within minutes). This is what turns a warm // no-change fetch into a no-op instead of a stat sweep. - let rootsQuietSince: ((sinceTs: number) => boolean) | null = null - void startRootWatchers().then(async (w) => { + let rootReuseValidation: ((sinceTs: number) => ParseReuseValidation) | null = null + // Mutable object properties keep cleanup visible to TypeScript even though + // setup assigns them from an asynchronous continuation. + const watcherLifecycle: { + state: RootWatcherState | null + resetValidator: (() => void) | null + } = { state: null, resetValidator: null } + const watcherSetup = startRootWatchers().then(async (w) => { + watcherLifecycle.state = w + if (!w) return const { setParseReuseValidator } = await import('./parser.js') // Clean means: the watchers were already armed when the parse happened, // and no filesystem event has landed since. lastEventAt of 0 is a quiet // system (clean for anything parsed after arming), not an unknown. - const quiet = (sinceTs: number): boolean => sinceTs >= w.startedAt && w.lastEventAt() < sinceTs - rootsQuietSince = quiet - setParseReuseValidator(quiet) - }).catch(() => { /* watcherless serve still works via the burst window */ }) + const validate = (sinceTs: number): ParseReuseValidation => classifyRootReuse(sinceTs, { + startedAt: w.startedAt, + lastEventAt: w.lastEventAt(), + healthy: w.healthy(), + }) + rootReuseValidation = validate + setParseReuseValidator(validate) + watcherLifecycle.resetValidator = () => setParseReuseValidator(null) + }).catch(() => { + watcherLifecycle.state?.close() + watcherLifecycle.state = null + }) // Output-level memo: an identical panel query while the roots are quiet // returns the previous stdout verbatim - the aggregation work is skipped @@ -197,7 +347,7 @@ export async function runStdioServe(buildProgram: () => Command): Promise // parse reuse; config.json is fingerprinted on every request because it can // change rendering without touching a provider root. const OUTPUT_MEMO_CAP_MS = 5 * 60 * 1000 - const outputMemo = new Map() + const outputMemo = new Map() let observedConfigFingerprint: string | null | undefined if (process.stdin.isTTY) { process.stderr.write('codeburn serve speaks JSON over stdio and exists for the desktop app to hold warm.\nNothing interactive happens here; press Ctrl+C to exit.\n') @@ -247,13 +397,14 @@ export async function runStdioServe(buildProgram: () => Command): Promise if ( configFingerprint !== null && memoHit?.configFingerprint === configFingerprint - && Date.now() - memoHit.at < OUTPUT_MEMO_CAP_MS - && rootsQuietSince?.(memoHit.at) + && Date.now() - memoHit.createdAt < OUTPUT_MEMO_CAP_MS + && rootReuseValidation?.(memoHit.validatedFrom) === 'clean' ) { write({ id: request.id, ok: true, output: memoHit.output }) return } try { + const parseStartedAt = Date.now() const { output, code } = await runCaptured( buildProgram, request.args, @@ -261,10 +412,10 @@ export async function runStdioServe(buildProgram: () => Command): Promise ) if (code === 0) { if (configFingerprint !== null) { - outputMemo.set(memoKey, { at: Date.now(), output, configFingerprint }) + outputMemo.set(memoKey, createOutputMemoEntry(parseStartedAt, Date.now(), output, configFingerprint)) } if (outputMemo.size > 32) { - const oldest = [...outputMemo.entries()].sort((a, b) => a[1].at - b[1].at)[0] + const oldest = [...outputMemo.entries()].sort((a, b) => a[1].createdAt - b[1].createdAt)[0] if (oldest) outputMemo.delete(oldest[0]) } write({ id: request.id, ok: true, output }) @@ -289,9 +440,24 @@ export async function runStdioServe(buildProgram: () => Command): Promise }) }) - // The app owns this process: stdin closing means the app is gone. - await new Promise((resolve) => { - rl.on('close', resolve) - process.stdin.on('end', resolve) + // The app owns this process: stdin closing (or failing) means the app is + // gone. Always release FSEvents handles and the module-global validator; + // otherwise an existing Claude root keeps a naturally closed child alive. + const transportClosed = new Promise((resolve) => { + rl.once('close', resolve) + process.stdin.once('end', resolve) + process.stdin.once('error', resolve) }) + try { + await transportClosed + } finally { + rl.close() + await watcherSetup + rootReuseValidation = null + try { + watcherLifecycle.resetValidator?.() + } finally { + watcherLifecycle.state?.close() + } + } } diff --git a/src/sync/ledger.ts b/src/sync/ledger.ts index b33f3588..8e81a63d 100644 --- a/src/sync/ledger.ts +++ b/src/sync/ledger.ts @@ -6,7 +6,7 @@ */ import { existsSync, mkdirSync, readFileSync, writeFileSync, unlinkSync, renameSync } from 'fs' -import { join } from 'path' +import { join, resolve } from 'path' import { getCodeburnCacheDir } from '../cache-dir.js' export interface LedgerEntry { @@ -17,15 +17,6 @@ export interface LedgerEntry { const SIX_MONTHS_MS = 180 * 24 * 60 * 60 * 1000 function ledgerCacheDir(): string { - const explicit = process.env.CODEBURN_CACHE_DIR - if (explicit?.trim()) return explicit - - // The sync ledger historically honored XDG_CACHE_HOME. Preserve that path - // so upgrades do not forget 180 days of sent keys and re-upload old calls; - // the ordinary CLI/desktop cache still shares the resolver below. - const xdg = process.env.XDG_CACHE_HOME - if (xdg?.trim()) return join(xdg, 'codeburn') - return getCodeburnCacheDir() } @@ -33,21 +24,59 @@ function ledgerPath(): string { return join(ledgerCacheDir(), 'sync-ledger.json') } -export function readLedger(): LedgerEntry[] { - const path = ledgerPath() - if (!existsSync(path)) return [] +// Before the shared cache resolver existed, sync alone wrote beneath +// XDG_CACHE_HOME. Treat that location as a one-time migration source only; +// CODEBURN_CACHE_DIR (when non-empty) is authoritative and must never import +// from an unrelated XDG tree. +function legacyXdgLedgerPath(): string | null { + if (process.env.CODEBURN_CACHE_DIR?.trim()) return null + const xdg = process.env.XDG_CACHE_HOME + if (!xdg?.trim()) return null + const legacy = join(xdg, 'codeburn', 'sync-ledger.json') + return resolve(legacy) === resolve(ledgerPath()) ? null : legacy +} + +function readLedgerFile(path: string): LedgerEntry[] | null { try { - const raw = readFileSync(path, 'utf-8') - const entries = JSON.parse(raw) as unknown - if (!Array.isArray(entries)) return [] + const entries = JSON.parse(readFileSync(path, 'utf-8')) as unknown + if (!Array.isArray(entries)) return null return entries.filter( (e): e is LedgerEntry => typeof e === 'object' && e !== null && typeof e.key === 'string' ) } catch { - return [] + return null } } +export function readLedger(): LedgerEntry[] { + const path = ledgerPath() + const legacyPath = legacyXdgLedgerPath() + const canonicalEntries = existsSync(path) ? readLedgerFile(path) : null + if (!legacyPath || !existsSync(legacyPath)) return canonicalEntries ?? [] + const legacyEntries = readLedgerFile(legacyPath) + if (!legacyEntries) return canonicalEntries ?? [] + + // Canonical wins for duplicate keys, but retain every key that exists only + // in the historical ledger so an upgrade cannot re-upload old calls. + const merged = [...(canonicalEntries ?? [])] + const keys = new Set(merged.map(entry => entry.key)) + for (const entry of legacyEntries) { + if (keys.has(entry.key)) continue + keys.add(entry.key) + merged.push(entry) + } + + // Publish the canonical copy before retiring the legacy source. If the + // write fails, keep and return the old ledger so deduplication still works. + try { + writeLedger(merged) + try { unlinkSync(legacyPath) } catch { /* canonical copy already wins */ } + } catch { + return merged + } + return merged +} + export function writeLedger(entries: LedgerEntry[]): void { const dir = ledgerCacheDir() mkdirSync(dir, { recursive: true }) @@ -83,11 +112,34 @@ export function ledgerKeySet(): Set { return new Set(readLedger().map(e => e.key)) } -/** Clear the ledger (for sync reset). Returns the number of entries removed. */ -export function clearLedger(): number { - const path = ledgerPath() - if (!existsSync(path)) return 0 - const count = readLedger().length - unlinkSync(path) - return count +function isMissingFileError(error: unknown): boolean { + return typeof error === 'object' && error !== null && 'code' in error && error.code === 'ENOENT' +} + +/** Clear every eligible ledger (for sync reset). Returns the number of unique + * entries removed. This deliberately bypasses readLedger(): reset must delete + * canonical and legacy files independently, never migrate one into the other. */ +export function clearLedger(): number { + const canonicalPath = ledgerPath() + const legacyPath = legacyXdgLedgerPath() + const targets = [canonicalPath, ...(legacyPath ? [legacyPath] : [])].map(path => ({ + path, + entries: readLedgerFile(path) ?? [], + })) + const removedKeys = new Set() + let deletionError: unknown + + // Attempt every target even if one unlink fails. A retry then has only the + // actual remainder to remove, while ENOENT is the idempotent success case. + for (const target of targets) { + try { + unlinkSync(target.path) + for (const entry of target.entries) removedKeys.add(entry.key) + } catch (error) { + if (!isMissingFileError(error) && deletionError === undefined) deletionError = error + } + } + + if (deletionError !== undefined) throw deletionError + return removedKeys.size } diff --git a/tests/cache-directory-switch.test.ts b/tests/cache-directory-switch.test.ts new file mode 100644 index 00000000..8b5bfbe0 --- /dev/null +++ b/tests/cache-directory-switch.test.ts @@ -0,0 +1,245 @@ +import { mkdir, mkdtemp, readFile, rm, stat, writeFile } from 'fs/promises' +import { existsSync } from 'fs' +import { tmpdir } from 'os' +import { join } from 'path' + +import { afterEach, beforeEach, describe, expect, it } from 'vitest' + +import { + fingerprintFile, + flushCodexCache, + readCachedCodexResults, + writeCachedCodexResults, +} from '../src/codex-cache.js' +import { + createAntigravityProvider, + flushAntigravityCache, +} from '../src/providers/antigravity.js' +import type { ParsedProviderCall } from '../src/providers/types.js' + +const originalCacheDir = process.env['CODEBURN_CACHE_DIR'] +const originalHome = process.env['HOME'] +const originalCodexHome = process.env['CODEX_HOME'] +let root: string + +function call(provider: string, marker: string): ParsedProviderCall { + return { + provider, + model: marker, + inputTokens: 1, + outputTokens: 1, + cacheCreationInputTokens: 0, + cacheReadInputTokens: 0, + cachedInputTokens: 0, + reasoningTokens: 0, + webSearchRequests: 0, + costUSD: 0, + tools: [], + bashCommands: [], + timestamp: '2026-08-12T00:00:00.000Z', + speed: 'standard', + deduplicationKey: `${provider}:${marker}`, + userMessage: '', + sessionId: marker, + } +} + +async function seedAntigravityCache( + cacheDir: string, + sourcePath: string, + marker: string, +): Promise { + const sourceStat = await stat(sourcePath) + await mkdir(cacheDir, { recursive: true }) + await writeFile(join(cacheDir, 'antigravity-results.json'), JSON.stringify({ + version: 5, + cascades: { + shared: { + mtimeMs: sourceStat.mtimeMs, + sizeBytes: sourceStat.size, + calls: [call('antigravity', marker)], + }, + }, + })) +} + +async function readAntigravityModel(sourcePath: string): Promise { + const parser = createAntigravityProvider().createSessionParser({ + path: sourcePath, + project: 'fixture', + provider: 'antigravity', + }, new Set()) + for await (const parsed of parser.parse()) return parsed.model + return undefined +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'codeburn-cache-switch-')) +}) + +afterEach(async () => { + if (originalCacheDir === undefined) delete process.env['CODEBURN_CACHE_DIR'] + else process.env['CODEBURN_CACHE_DIR'] = originalCacheDir + if (originalHome === undefined) delete process.env['HOME'] + else process.env['HOME'] = originalHome + if (originalCodexHome === undefined) delete process.env['CODEX_HOME'] + else process.env['CODEX_HOME'] = originalCodexHome + await rm(root, { recursive: true, force: true }) +}) + +describe('call-time CODEBURN_CACHE_DIR isolation', () => { + it('keeps Codex reads and writes keyed by the active cache directory', async () => { + const sourcePath = join(root, 'rollout.jsonl') + const cacheA = join(root, 'cache-a') + const cacheB = join(root, 'cache-b') + await writeFile(sourcePath, '{}\n') + const fingerprint = await fingerprintFile(sourcePath) + expect(fingerprint).not.toBeNull() + + process.env['CODEBURN_CACHE_DIR'] = cacheA + await writeCachedCodexResults(sourcePath, 'project-a', [call('codex', 'from-a')], fingerprint!) + await flushCodexCache() + + process.env['CODEBURN_CACHE_DIR'] = cacheB + expect(await readCachedCodexResults(sourcePath)).toBeNull() + await writeCachedCodexResults(sourcePath, 'project-b', [call('codex', 'from-b')], fingerprint!) + await flushCodexCache() + + const diskB = JSON.parse(await readFile(join(cacheB, 'codex-results.json'), 'utf8')) + expect(diskB.files[sourcePath].calls.map((entry: ParsedProviderCall) => entry.model)).toEqual(['from-b']) + + process.env['CODEBURN_CACHE_DIR'] = cacheA + expect((await readCachedCodexResults(sourcePath))?.map(entry => entry.model)).toEqual(['from-a']) + }) + + it('does not flush dirty Codex state from A into B', async () => { + const sourceA = join(root, 'a.jsonl') + const sourceB = join(root, 'b.jsonl') + const cacheA = join(root, 'cache-a-dirty') + const cacheB = join(root, 'cache-b-dirty') + await writeFile(sourceA, 'a\n') + await writeFile(sourceB, 'b\n') + + process.env['CODEBURN_CACHE_DIR'] = cacheA + await writeCachedCodexResults(sourceA, 'project-a', [call('codex', 'dirty-a')], (await fingerprintFile(sourceA))!) + + process.env['CODEBURN_CACHE_DIR'] = cacheB + await writeCachedCodexResults(sourceB, 'project-b', [call('codex', 'dirty-b')], (await fingerprintFile(sourceB))!) + await flushCodexCache() + + const diskB = JSON.parse(await readFile(join(cacheB, 'codex-results.json'), 'utf8')) + expect(Object.keys(diskB.files)).toEqual([sourceB]) + + process.env['CODEBURN_CACHE_DIR'] = cacheA + await flushCodexCache() + const diskA = JSON.parse(await readFile(join(cacheA, 'codex-results.json'), 'utf8')) + expect(Object.keys(diskA.files)).toEqual([sourceA]) + }) + + it('pins Codex reads, dirty writes, and flushes to the parse call-time directory', async () => { + const home = join(root, 'parse-home') + const codexHome = join(root, 'parse-codex-home') + const sessionDir = join(codexHome, 'sessions', '2026', '08', '12') + const cacheA = join(root, 'parse-cache-a') + const cacheB = join(root, 'parse-cache-b') + await mkdir(sessionDir, { recursive: true }) + await mkdir(home, { recursive: true }) + const sourcePath = join(sessionDir, 'rollout-cache-dir-switch.jsonl') + await writeFile(sourcePath, [ + JSON.stringify({ + type: 'session_meta', + timestamp: '2026-08-12T10:00:00.000Z', + payload: { + cwd: '/Users/test/cache-dir-transaction', + originator: 'codex-cli', + session_id: 'cache-dir-transaction', + model: 'gpt-5.3-codex', + }, + }), + JSON.stringify({ + type: 'event_msg', + timestamp: '2026-08-12T10:01:00.000Z', + payload: { + type: 'token_count', + info: { + model: 'gpt-5.3-codex', + last_token_usage: { + input_tokens: 10, + cached_input_tokens: 0, + output_tokens: 5, + reasoning_output_tokens: 0, + total_tokens: 15, + }, + total_token_usage: { + input_tokens: 10, + cached_input_tokens: 0, + output_tokens: 5, + reasoning_output_tokens: 0, + total_tokens: 15, + }, + }, + }, + }), + ].join('\n') + '\n') + + process.env['HOME'] = home + process.env['CODEX_HOME'] = codexHome + process.env['CODEBURN_CACHE_DIR'] = cacheA + const { clearSessionCache, parseAllSessions } = await import('../src/parser.js') + clearSessionCache() + + // parseAllSessions reaches its first await before any Codex cache access. + // Switching the host env immediately after invocation deterministically + // exercises every later read/write/flush under the captured A transaction. + const parsing = parseAllSessions(undefined, 'codex') + process.env['CODEBURN_CACHE_DIR'] = cacheB + const projects = await parsing + + expect(projects.some(project => project.sessions.some(session => + session.turns.some(turn => turn.assistantCalls.some(entry => entry.provider === 'codex')) + ))).toBe(true) + expect(existsSync(join(cacheA, 'codex-results.json'))).toBe(true) + expect(existsSync(join(cacheB, 'codex-results.json'))).toBe(false) + const diskA = JSON.parse(await readFile(join(cacheA, 'codex-results.json'), 'utf8')) + expect(diskA.files[sourcePath].calls).toHaveLength(1) + clearSessionCache() + }) + + it('loads Antigravity cache entries from the active directory after A to B', async () => { + const sourcePath = join(root, 'shared.pb') + const cacheA = join(root, 'agy-cache-a') + const cacheB = join(root, 'agy-cache-b') + await writeFile(sourcePath, 'fixture') + await seedAntigravityCache(cacheA, sourcePath, 'from-a') + await seedAntigravityCache(cacheB, sourcePath, 'from-b') + + process.env['CODEBURN_CACHE_DIR'] = cacheA + expect(await readAntigravityModel(sourcePath)).toBe('from-a') + + process.env['CODEBURN_CACHE_DIR'] = cacheB + expect(await readAntigravityModel(sourcePath)).toBe('from-b') + }) + + it('does not flush dirty Antigravity state from A into B', async () => { + const sourcePath = join(root, 'shared.pb') + const cacheA = join(root, 'agy-cache-a-dirty') + const cacheB = join(root, 'agy-cache-b-dirty') + await writeFile(sourcePath, 'fixture') + await seedAntigravityCache(cacheA, sourcePath, 'from-a') + await seedAntigravityCache(cacheB, sourcePath, 'from-b') + + process.env['CODEBURN_CACHE_DIR'] = cacheA + expect(await readAntigravityModel(sourcePath)).toBe('from-a') + + // The provider parse transaction captures A. Even if the host changes its + // call-time env before the deferred flush, eviction/publication stays on A. + process.env['CODEBURN_CACHE_DIR'] = cacheB + await flushAntigravityCache(new Set(), cacheA) + + expect(existsSync(join(cacheB, 'antigravity-results.json'))).toBe(true) + const diskB = JSON.parse(await readFile(join(cacheB, 'antigravity-results.json'), 'utf8')) + expect(diskB.cascades.shared.calls[0].model).toBe('from-b') + const diskA = JSON.parse(await readFile(join(cacheA, 'antigravity-results.json'), 'utf8')) + expect(diskA.cascades).toEqual({}) + }) +}) diff --git a/tests/parser.test.ts b/tests/parser.test.ts index 4bd0c5c2..ff029bc0 100644 --- a/tests/parser.test.ts +++ b/tests/parser.test.ts @@ -23,6 +23,7 @@ import type { SessionSource, SessionParser, ParsedProviderCall } from '../src/pr let _synthSources: SessionSource[] = [] let _synthDurable = false let _synthYields: ParsedProviderCall[] = [] +let _synthOnParse: (() => void | Promise) | null = null vi.mock('../src/providers/index.js', async (importOriginal) => { type Mod = typeof import('../src/providers/index.js') @@ -52,6 +53,7 @@ vi.mock('../src/providers/index.js', async (importOriginal) => { createSessionParser(_s: SessionSource, _k: Set): SessionParser { return { async *parse(): AsyncGenerator { + await _synthOnParse?.() for (const call of _synthYields) { // Respect seenKeys so that when multiple sources share the same // dedup key, only the first source yields it (mirrors real parsers). @@ -190,13 +192,16 @@ beforeEach(async () => { _synthSources = [] _synthDurable = false _synthYields = [] + _synthOnParse = null }) afterEach(async () => { clearSessionCache() + setParseReuseValidator(null) vi.unstubAllEnvs() _synthSources = [] + _synthOnParse = null await rm(tmpHome, { recursive: true, force: true }) await rm(tmpCache, { recursive: true, force: true }) @@ -728,6 +733,75 @@ describe('(q) parse burst reuse (CODEBURN_PARSE_BURST_MS)', () => { }) describe('(r) validated parse reuse (setParseReuseValidator)', () => { + it('falls back to the exact TTL when watcher coverage is unknown, but rejects dirty', async () => { + vi.stubEnv('CODEBURN_PARSE_BURST_MS', '0') + clearSessionCache() + const start = new Date(Date.now() - 60 * 60 * 1000) + const end = new Date() + const ts = new Date(Date.now() - 10 * 60 * 1000).toISOString() + const synthFile = join(tmpHome, 'synth-unknown-exact.txt') + await writeFile(synthFile, 'first input') + _synthSources = [{ path: synthFile, project: 'p', provider: 'test-synthetic' }] + _synthYields = [{ + provider: 'test-synthetic', model: 'synth-model', + inputTokens: 1, outputTokens: 5, cacheCreationInputTokens: 0, cacheReadInputTokens: 0, + cachedInputTokens: 0, reasoningTokens: 0, webSearchRequests: 0, + costUSD: 0, costIsEstimated: false, tools: [], bashCommands: [], skills: [], + timestamp: ts, speed: 'standard', deduplicationKey: 'synth-unknown-exact-1', userMessage: 'hi', sessionId: 'sue-1', + }] as never + + expect(totalOutput(await parseAllSessions({ start, end }, 'test-synthetic'))).toBe(5) + _synthYields = [..._synthYields, { + ...( _synthYields[0] as object ), deduplicationKey: 'synth-unknown-exact-2', outputTokens: 7, + }] as never + await writeFile(synthFile, 'second input with changed fingerprint') + + // An unhealthy/pre-arm watcher cannot extend freshness, but it must retain + // the normal exact-key TTL instead of forcing a full rescan every request. + setParseReuseValidator(() => 'unknown') + expect(totalOutput(await parseAllSessions({ start, end }, 'test-synthetic'))).toBe(5) + + // The same entry must be rejected immediately once a real change is known. + setParseReuseValidator(() => 'dirty') + expect(totalOutput(await parseAllSessions({ start, end }, 'test-synthetic'))).toBe(12) + }) + + it('falls back to the short burst when watcher coverage is unknown, but dirty wins inside it', async () => { + vi.stubEnv('CODEBURN_PARSE_BURST_MS', '10000') + clearSessionCache() + const start = new Date(Date.now() - 60 * 60 * 1000) + const firstEnd = new Date() + const ts = new Date(Date.now() - 10 * 60 * 1000).toISOString() + const synthFile = join(tmpHome, 'synth-unknown-burst.txt') + await writeFile(synthFile, 'first input') + _synthSources = [{ path: synthFile, project: 'p', provider: 'test-synthetic' }] + _synthYields = [{ + provider: 'test-synthetic', model: 'synth-model', + inputTokens: 1, outputTokens: 5, cacheCreationInputTokens: 0, cacheReadInputTokens: 0, + cachedInputTokens: 0, reasoningTokens: 0, webSearchRequests: 0, + costUSD: 0, costIsEstimated: false, tools: [], bashCommands: [], skills: [], + timestamp: ts, speed: 'standard', deduplicationKey: 'synth-unknown-burst-1', userMessage: 'hi', sessionId: 'sub-1', + }] as never + + expect(totalOutput(await parseAllSessions({ start, end: firstEnd }, 'test-synthetic'))).toBe(5) + _synthYields = [..._synthYields, { + ...( _synthYields[0] as object ), deduplicationKey: 'synth-unknown-burst-2', outputTokens: 7, + }] as never + await writeFile(synthFile, 'second input with changed fingerprint') + + setParseReuseValidator(() => 'unknown') + expect(totalOutput(await parseAllSessions( + { start, end: new Date(firstEnd.getTime() + 100) }, + 'test-synthetic', + ))).toBe(5) + + setParseReuseValidator(() => 'dirty') + expect(totalOutput(await parseAllSessions( + { start, end: new Date(firstEnd.getTime() + 200) }, + 'test-synthetic', + ))).toBe(12) + }) + it('reuses past the burst window while the validator reports quiet, never when dirty', async () => { vi.stubEnv('CODEBURN_PARSE_BURST_MS', '1') clearSessionCache() @@ -750,14 +824,14 @@ describe('(r) validated parse reuse (setParseReuseValidator)', () => { // 1ms burst window has certainly elapsed; with a quiet validator the // previous parse is still served (world changed, result must not). await new Promise(r => setTimeout(r, 5)) - setParseReuseValidator(() => true) + setParseReuseValidator(() => 'clean') _synthYields = [..._synthYields, { ...( _synthYields[0] as object ), deduplicationKey: 'synth-val-2', outputTokens: 7 }] as never await writeFile(synthFile, 'placeholder v2') const second = await parseAllSessions({ start, end: new Date(Date.now() + 500) }, 'test-synthetic') expect(totalOutput(second)).toBe(5) // A dirty validator ends the reuse: fresh parse sees the new call. - setParseReuseValidator(() => false) + setParseReuseValidator(() => 'dirty') const third = await parseAllSessions({ start, end: new Date(Date.now() + 1000) }, 'test-synthetic') expect(totalOutput(third)).toBe(12) @@ -766,4 +840,85 @@ describe('(r) validated parse reuse (setParseReuseValidator)', () => { _synthSources = [] _synthYields = [] }) + + it('rejects an exact-key memo when a root event arrived during its parse', async () => { + clearSessionCache() + const start = new Date(Date.now() - 60 * 60 * 1000) + const end = new Date() + const ts = new Date(Date.now() - 10 * 60 * 1000).toISOString() + const synthFile = join(tmpHome, 'synth-exact-event-during-parse.txt') + await writeFile(synthFile, 'first input') + _synthSources = [{ path: synthFile, project: 'p', provider: 'test-synthetic' }] + _synthYields = [{ + provider: 'test-synthetic', model: 'synth-model', + inputTokens: 1, outputTokens: 5, cacheCreationInputTokens: 0, cacheReadInputTokens: 0, + cachedInputTokens: 0, reasoningTokens: 0, webSearchRequests: 0, + costUSD: 0, costIsEstimated: false, tools: [], bashCommands: [], skills: [], + timestamp: ts, speed: 'standard', deduplicationKey: 'synth-exact-event-1', userMessage: 'hi', sessionId: 'see-1', + }] as never + + let rootEventAt = 0 + setParseReuseValidator(sinceTs => rootEventAt === 0 || rootEventAt < sinceTs ? 'clean' : 'dirty') + _synthOnParse = async () => { + await new Promise(resolve => setTimeout(resolve, 10)) + rootEventAt = Date.now() + await new Promise(resolve => setTimeout(resolve, 10)) + } + const first = await parseAllSessions({ start, end }, 'test-synthetic') + expect(totalOutput(first)).toBe(5) + _synthOnParse = null + + _synthYields = [..._synthYields, { + ...( _synthYields[0] as object ), deduplicationKey: 'synth-exact-event-2', outputTokens: 7, + }] as never + await writeFile(synthFile, 'second input') + const second = await parseAllSessions({ start, end }, 'test-synthetic') + expect(totalOutput(second)).toBe(12) + }) + + it('does not bless a root event that arrived while the cached parse was running', async () => { + vi.stubEnv('CODEBURN_PARSE_BURST_MS', '1') + clearSessionCache() + const start = new Date(Date.now() - 60 * 60 * 1000) + const firstEnd = new Date() + const ts = new Date(Date.now() - 10 * 60 * 1000).toISOString() + const synthFile = join(tmpHome, 'synth-event-during-parse.txt') + await writeFile(synthFile, 'first input') + _synthSources = [{ path: synthFile, project: 'p', provider: 'test-synthetic' }] + _synthYields = [{ + provider: 'test-synthetic', model: 'synth-model', + inputTokens: 1, outputTokens: 5, cacheCreationInputTokens: 0, cacheReadInputTokens: 0, + cachedInputTokens: 0, reasoningTokens: 0, webSearchRequests: 0, + costUSD: 0, costIsEstimated: false, tools: [], bashCommands: [], skills: [], + timestamp: ts, speed: 'standard', deduplicationKey: 'synth-event-1', userMessage: 'hi', sessionId: 'se-1', + }] as never + + let rootEventAt = 0 + _synthOnParse = async () => { + // Bracket the controlled event so it is strictly after parse start and + // strictly before completion, independent of same-millisecond clocks. + await new Promise(resolve => setTimeout(resolve, 10)) + rootEventAt = Date.now() + await new Promise(resolve => setTimeout(resolve, 10)) + } + const first = await parseAllSessions({ start, end: firstEnd }, 'test-synthetic') + expect(totalOutput(first)).toBe(5) + expect(rootEventAt).toBeGreaterThan(0) + _synthOnParse = null + + // Outside the 1ms burst, old code validated against cachePut completion + // and reused stale output because the in-parse event appeared older. The + // parse-start timestamp makes the validator reject reuse and rescan. + await new Promise(resolve => setTimeout(resolve, 5)) + setParseReuseValidator(sinceTs => rootEventAt < sinceTs ? 'clean' : 'dirty') + _synthYields = [..._synthYields, { + ...( _synthYields[0] as object ), deduplicationKey: 'synth-event-2', outputTokens: 7, + }] as never + await writeFile(synthFile, 'second input') + const second = await parseAllSessions( + { start, end: new Date(firstEnd.getTime() + 500) }, + 'test-synthetic', + ) + expect(totalOutput(second)).toBe(12) + }) }) diff --git a/tests/providers/claude-config-dirs.test.ts b/tests/providers/claude-config-dirs.test.ts index 571469b1..899fb335 100644 --- a/tests/providers/claude-config-dirs.test.ts +++ b/tests/providers/claude-config-dirs.test.ts @@ -390,6 +390,37 @@ describe('claude provider — config.json claudeConfigDirs (menubar-driven)', () expect(paths).toContain(join(personal, 'projects', '-Users-you-app')) }) + it('invalidates the exact parse memo when config.json adds a Claude discovery root', async () => { + const work = await makeConfigDir('claude-work', []) + const personal = await makeConfigDir('claude-personal', []) + const slug = '-Users-you-shared-app' + const cwd = '/Users/you/shared-app' + await writeSession(work, slug, 'sess-work', [ + summaryLine('sess-work', cwd), + userLine('u1', 'sess-work', cwd, 'hi from work'), + assistantLine('a1', 'u1', 'sess-work', cwd), + ]) + await writeSession(personal, slug, 'sess-personal', [ + summaryLine('sess-personal', cwd), + userLine('u2', 'sess-personal', cwd, 'hi from personal'), + assistantLine('a2', 'u2', 'sess-personal', cwd), + ]) + + await writeConfigJson([work]) + const first = await parseAllSessions(undefined, 'claude') + expect(first.flatMap(project => project.sessions).map(session => session.sessionId)).toEqual(['sess-work']) + + // Same argv/date range and unchanged env: only the effective roots sourced + // from config.json differ. A resident process must not return the exact-key + // memo populated by the first call. + await writeConfigJson([work, personal]) + const second = await parseAllSessions(undefined, 'claude') + expect(second.flatMap(project => project.sessions).map(session => session.sessionId).sort()).toEqual([ + 'sess-personal', + 'sess-work', + ]) + }) + it('lets env CLAUDE_CONFIG_DIRS override config.json', async () => { const fromEnv = await makeConfigDir('claude-env', ['-Users-you-app']) const fromFile = await makeConfigDir('claude-file', ['-Users-you-app']) diff --git a/tests/serve-stdio.test.ts b/tests/serve-stdio.test.ts index 427b0950..bbc230fe 100644 --- a/tests/serve-stdio.test.ts +++ b/tests/serve-stdio.test.ts @@ -1,7 +1,31 @@ import { describe, it, expect, beforeAll, afterAll } from 'vitest' import { spawn, type ChildProcess } from 'child_process' -import { mkdir, writeFile } from 'fs/promises' +import { mkdir, readFile, writeFile } from 'fs/promises' import { join } from 'path' +import { classifyRootReuse, createOutputMemoEntry } from '../src/serve.js' + +it('timestamps a completed output memo before parsing begins', () => { + const parseStartedAt = 100 + const rootEventDuringParseAt = 150 + const parseCompletedAt = 200 + const memo = createOutputMemoEntry(parseStartedAt, parseCompletedAt, 'output', 'config') + const rootsQuietSince = (sinceTs: number): boolean => rootEventDuringParseAt < sinceTs + + // The old completion timestamp incorrectly made the in-parse event look + // older than the memo. The start timestamp keeps it visible to validation. + expect(rootsQuietSince(parseCompletedAt)).toBe(true) + expect(memo.createdAt).toBe(parseCompletedAt) + expect(memo.validatedFrom).toBe(parseStartedAt) + expect(rootsQuietSince(memo.validatedFrom)).toBe(false) +}) + +it('classifies watcher gaps as unknown without confusing them with dirty roots', () => { + expect(classifyRootReuse(100, { startedAt: 50, lastEventAt: 0, healthy: false })).toBe('unknown') + expect(classifyRootReuse(100, { startedAt: 150, lastEventAt: 0, healthy: true })).toBe('unknown') + expect(classifyRootReuse(100, { startedAt: 50, lastEventAt: 100, healthy: false })).toBe('dirty') + expect(classifyRootReuse(100, { startedAt: 50, lastEventAt: 100, healthy: true })).toBe('dirty') + expect(classifyRootReuse(100, { startedAt: 50, lastEventAt: 99, healthy: true })).toBe('clean') +}) // End-to-end protocol test for `codeburn serve --stdio` (the desktop app's // resident query server). Runs the real entry through tsx against the @@ -31,6 +55,9 @@ describe('codeburn serve --stdio', () => { const home = process.env['HOME']! configPath = join(home, '.config', 'codeburn', 'config.json') await mkdir(join(home, '.config', 'codeburn'), { recursive: true }) + // Give the resident process one real provider root to arm. With no + // successfully armed roots, event-driven reuse correctly stays disabled. + await mkdir(join(home, '.claude', 'projects'), { recursive: true }) await writeFile(configPath, JSON.stringify({ currency: { code: 'USD' } }), 'utf8') // Keep the EUR half of the config-freshness regression fully offline. @@ -106,6 +133,51 @@ describe('codeburn serve --stdio', () => { expect(res['refused']).toBe(true) }) + it('refuses every optimize apply-only option without touching shell config or the action journal', async () => { + const home = process.env['HOME']! + const zshrc = join(home, '.zshrc') + const journal = join(home, '.config', 'codeburn', 'actions', 'journal.jsonl') + await writeFile(zshrc, '# user-owned\n', 'utf8') + + // `optimize` is the only served command whose Commander definition also + // has mutation-capable options. The full request below used to execute a + // shell-config action inside the resident process. + const applied = await request(300, [ + 'optimize', '--apply', '--yes', '--only', 'bash-output-cap', '--period', 'today', + ]) + expect(applied).toMatchObject({ ok: false, refused: true }) + + // Keep the allowlist categorical: apply-only modifiers are not useful to + // a read query and must not become resident options on their own either. + for (const [id, args] of [ + [301, ['optimize', '--yes']], + [302, ['optimize', '--dry-run']], + [303, ['optimize', '--only', 'bash-output-cap']], + ] as const) { + expect(await request(id, [...args])).toMatchObject({ ok: false, refused: true }) + } + + expect(await readFile(zshrc, 'utf8')).toBe('# user-owned\n') + await expect(readFile(journal, 'utf8')).rejects.toMatchObject({ code: 'ENOENT' }) + }, 60_000) + + it('accepts the reviewed read-only option surface for every served command', async () => { + const commands: Array<[number, string[]]> = [ + [310, ['status', '--format', 'json', '--period', 'today']], + [311, ['overview', '--period', 'today', '--no-color']], + [312, ['models', '--format', 'json', '--period', 'today', '--no-totals']], + [313, ['sessions', '--format', 'json', '--period', 'today', '--no-pager']], + [314, ['compare', '--format', 'json', '--period', 'today']], + [315, ['yield', '--format', 'json', '--period', 'today']], + [316, ['spend', '--format', 'flow-json', '--period', 'today']], + [317, ['optimize', '--format', 'json', '--period', 'today']], + [318, ['audit', '--format', 'json', '--period', 'today']], + ] + for (const [id, args] of commands) { + expect(await request(id, args)).toMatchObject({ ok: true }) + } + }, 60_000) + it('survives a malformed request line and keeps serving', async () => { sendRaw('this is not json') const res = await request(6, ['status', '--format', 'menubar-json', '--period', 'today']) @@ -113,13 +185,68 @@ describe('codeburn serve --stdio', () => { }, 60_000) it('streams captured command stderr as protocol progress frames', async () => { - const res = await request(7, ['status', '--definitely-not-a-real-option']) + const res = await request(7, ['status', '--provider', 'definitely-not-a-real-provider']) expect(res['ok']).toBe(false) const frames = progressFrames.get(7) ?? [] expect(frames.length).toBeGreaterThan(0) expect(frames.every(frame => Object.keys(frame).sort().join(',') === 'id,progress')).toBe(true) - expect(frames.map(frame => frame['progress']).join('')).toContain('unknown option') + expect(frames.map(frame => frame['progress']).join('')).toContain('unknown provider') + }, 60_000) + + it('discovers a newly configured Claude root on identical resident argv', async () => { + const home = process.env['HOME']! + const rootA = join(home, 'claude-root-a') + const rootB = join(home, 'claude-root-b') + const slug = '-Users-test-shared-project' + const cwd = '/Users/test/shared-project' + + const writeClaudeSession = async (root: string, sessionId: string, marker: string): Promise => { + const projectDir = join(root, 'projects', slug) + await mkdir(projectDir, { recursive: true }) + const lines = [ + { + type: 'summary', summary: marker, leafUuid: `leaf-${marker}`, sessionId, cwd, + timestamp: '2026-08-12T10:00:00.000Z', + }, + { + type: 'user', uuid: `user-${marker}`, sessionId, cwd, + timestamp: '2026-08-12T10:00:01.000Z', message: { role: 'user', content: marker }, + }, + { + type: 'assistant', uuid: `assistant-${marker}`, parentUuid: `user-${marker}`, sessionId, cwd, + timestamp: '2026-08-12T10:00:02.000Z', + message: { + id: `msg-${marker}`, type: 'message', role: 'assistant', model: 'claude-sonnet-4-6', + content: [{ type: 'text', text: 'reply' }], usage: { input_tokens: 100, output_tokens: 50 }, + }, + }, + ] + await writeFile(join(projectDir, `${sessionId}.jsonl`), lines.map(line => JSON.stringify(line)).join('\n')) + } + + await writeClaudeSession(rootA, 'resident-session-a', 'a') + await writeClaudeSession(rootB, 'resident-session-b', 'b') + const args = ['sessions', '--period', 'lifetime', '--provider', 'claude', '--format', 'json', '--no-pager'] + + await writeFile(configPath, JSON.stringify({ claudeConfigDirs: [rootA] }), 'utf8') + const first = await request(200, args) + expect(first['ok']).toBe(true) + expect((JSON.parse(first['output'] as string) as Array<{ sessionId: string }>).map(row => row.sessionId)).toEqual([ + 'resident-session-a', + ]) + + // Same command in the same process; only config.json adds root B. + await writeFile(configPath, JSON.stringify({ claudeConfigDirs: [rootA, rootB] }), 'utf8') + const second = await request(201, args) + expect(second['ok']).toBe(true) + expect((JSON.parse(second['output'] as string) as Array<{ sessionId: string }>).map(row => row.sessionId).sort()).toEqual([ + 'resident-session-a', + 'resident-session-b', + ]) + + // Keep the following currency-freshness regression self-contained. + await writeFile(configPath, JSON.stringify({ currency: { code: 'USD' } }), 'utf8') }, 60_000) it('invalidates identical-argv output memo immediately when config.json changes', async () => { @@ -172,4 +299,47 @@ describe('codeburn serve --stdio', () => { currency: { code: string; rate: number } }).currency).toMatchObject({ code: 'USD', rate: 1 }) }, 60_000) + + it('exits on natural stdin EOF after arming a watcher for an existing Claude root', async () => { + const claudeRoot = join(process.env['HOME']!, 'claude-eof-root') + await mkdir(join(claudeRoot, 'projects'), { recursive: true }) + + const eofChild = spawn(process.execPath, ['--import', 'tsx', join(__dirname, '..', 'src', 'cli.ts'), 'serve', '--stdio'], { + stdio: ['pipe', 'pipe', 'ignore'], + env: { ...process.env, CLAUDE_CONFIG_DIR: claudeRoot }, + }) + let stdout = '' + const becameReady = new Promise((resolve, reject) => { + eofChild.once('error', reject) + eofChild.stdout!.setEncoding('utf8') + eofChild.stdout!.on('data', (chunk: string) => { + stdout += chunk + if (stdout.split('\n').some(line => { + try { return (JSON.parse(line) as { ready?: boolean }).ready === true } catch { return false } + })) resolve() + }) + eofChild.once('exit', (code, signal) => reject(new Error(`serve exited before ready: ${code ?? signal}`))) + }) + const exited = new Promise(resolve => eofChild.once('exit', () => resolve(true))) + + let naturalExit = false + try { + await becameReady + // READY is intentionally emitted before provider probing; give the real + // watcher setup time to finish so the regression exercises its handle. + await new Promise(resolve => setTimeout(resolve, 500)) + eofChild.stdin!.end() + naturalExit = await Promise.race([ + exited, + new Promise(resolve => setTimeout(() => resolve(false), 2_000)), + ]) + } finally { + if (!naturalExit) { + eofChild.kill('SIGKILL') + await exited + } + } + + expect(naturalExit).toBe(true) + }, 10_000) }) diff --git a/tests/sync-ledger-otlp.test.ts b/tests/sync-ledger-otlp.test.ts index 08de4ac5..b7c72ca6 100644 --- a/tests/sync-ledger-otlp.test.ts +++ b/tests/sync-ledger-otlp.test.ts @@ -2,7 +2,7 @@ * Unit tests for sync ledger and OTLP payload builder. */ -import { describe, it, expect, beforeEach, afterEach } from 'vitest' +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest' import { mkdtemp, rm } from 'fs/promises' import { join } from 'path' import { tmpdir } from 'os' @@ -312,6 +312,91 @@ describe('ledger', () => { expect(clearLedger()).toBe(0) }) + it('clearLedger removes coexisting canonical and eligible legacy ledgers without adopting either', async () => { + const { clearLedger } = await import('../src/sync/ledger.js') + const { existsSync, mkdirSync, writeFileSync } = await import('fs') + const canonicalDir = join(tmpDir, '.cache', 'codeburn') + const xdgDir = join(tmpDir, 'xdg-clear-coexisting') + const legacyDir = join(xdgDir, 'codeburn') + + delete process.env.CODEBURN_CACHE_DIR + process.env.XDG_CACHE_HOME = xdgDir + mkdirSync(canonicalDir, { recursive: true }) + mkdirSync(legacyDir, { recursive: true }) + writeFileSync(join(canonicalDir, 'sync-ledger.json'), JSON.stringify([ + { key: 'canonical', ts: '2026-07-02T00:00:00Z' }, + { key: 'duplicate', ts: '2026-07-03T00:00:00Z' }, + ])) + writeFileSync(join(legacyDir, 'sync-ledger.json'), JSON.stringify([ + { key: 'legacy', ts: '2026-07-01T00:00:00Z' }, + { key: 'duplicate', ts: '2025-01-01T00:00:00Z' }, + ])) + + expect(clearLedger()).toBe(3) + expect(existsSync(join(canonicalDir, 'sync-ledger.json'))).toBe(false) + expect(existsSync(join(legacyDir, 'sync-ledger.json'))).toBe(false) + }) + + it('clearLedger attempts both targets, reports a real unlink failure, and can retry the remainder', async () => { + const fs = await import('fs') + const canonicalDir = join(tmpDir, '.cache', 'codeburn') + const canonicalPath = join(canonicalDir, 'sync-ledger.json') + const xdgDir = join(tmpDir, 'xdg-clear-retry') + const legacyDir = join(xdgDir, 'codeburn') + const legacyPath = join(legacyDir, 'sync-ledger.json') + + delete process.env.CODEBURN_CACHE_DIR + process.env.XDG_CACHE_HOME = xdgDir + fs.mkdirSync(canonicalDir, { recursive: true }) + fs.mkdirSync(legacyDir, { recursive: true }) + fs.writeFileSync(canonicalPath, JSON.stringify([ + { key: 'canonical', ts: '2026-07-02T00:00:00Z' }, + ])) + fs.writeFileSync(legacyPath, JSON.stringify([ + { key: 'legacy', ts: '2026-07-01T00:00:00Z' }, + ])) + + const attempts: string[] = [] + let failCanonicalOnce = true + vi.doMock('fs', async () => { + const actual = await vi.importActual('fs') + return { + ...actual, + unlinkSync: (path: fs.PathLike) => { + const value = String(path) + attempts.push(value) + if (value === canonicalPath && failCanonicalOnce) { + failCanonicalOnce = false + throw Object.assign(new Error('injected canonical unlink failure'), { code: 'EACCES' }) + } + return actual.unlinkSync(path) + }, + } + }) + vi.resetModules() + + try { + const { clearLedger } = await import('../src/sync/ledger.js') + expect(() => clearLedger()).toThrow('injected canonical unlink failure') + expect(attempts).toContain(canonicalPath) + expect(attempts).toContain(legacyPath) + expect(fs.existsSync(canonicalPath)).toBe(true) + expect(fs.existsSync(legacyPath)).toBe(false) + expect(JSON.parse(fs.readFileSync(canonicalPath, 'utf8'))).toEqual([ + { key: 'canonical', ts: '2026-07-02T00:00:00Z' }, + ]) + + // The successful legacy deletion is not replayed or migrated. Retrying + // removes only the canonical remainder; its missing peer is ENOENT-safe. + expect(clearLedger()).toBe(1) + expect(fs.existsSync(canonicalPath)).toBe(false) + expect(fs.existsSync(legacyPath)).toBe(false) + } finally { + vi.doUnmock('fs') + vi.resetModules() + } + }) + it('corrupt ledger file reads as empty (crash-safe recovery)', async () => { const { readLedger } = await import('../src/sync/ledger.js') const { mkdirSync, writeFileSync } = await import('fs') @@ -364,17 +449,83 @@ describe('ledger', () => { expect(existsSync(join(tmpDir, '.cache', 'codeburn', 'sync-ledger.json'))).toBe(true) }) - it('uses XDG_CACHE_HOME/codeburn when the explicit override is absent', async () => { - const { writeLedger } = await import('../src/sync/ledger.js') - const { existsSync } = await import('fs') + it('adopts an XDG-only legacy ledger into the canonical default and writes there thereafter', async () => { + const { appendToLedger, readLedger } = await import('../src/sync/ledger.js') + const { existsSync, mkdirSync, readFileSync, writeFileSync } = await import('fs') const { join } = await import('path') const xdgDir = join(tmpDir, 'xdg-cache') + const legacyDir = join(xdgDir, 'codeburn') + const canonicalDir = join(tmpDir, '.cache', 'codeburn') delete process.env.CODEBURN_CACHE_DIR process.env.XDG_CACHE_HOME = xdgDir - writeLedger([{ key: 'xdg', ts: '2026-07-01T00:00:00Z' }]) + mkdirSync(legacyDir, { recursive: true }) + writeFileSync(join(legacyDir, 'sync-ledger.json'), JSON.stringify([ + { key: 'legacy', ts: '2026-07-01T00:00:00Z' }, + ])) - expect(existsSync(join(xdgDir, 'codeburn', 'sync-ledger.json'))).toBe(true) + expect(readLedger().map(entry => entry.key)).toEqual(['legacy']) + expect(existsSync(join(canonicalDir, 'sync-ledger.json'))).toBe(true) + expect(existsSync(join(legacyDir, 'sync-ledger.json'))).toBe(false) + + appendToLedger([{ key: 'canonical', ts: '2026-07-02T00:00:00Z' }]) + expect(JSON.parse(readFileSync(join(canonicalDir, 'sync-ledger.json'), 'utf8')).map((entry: { key: string }) => entry.key)).toEqual([ + 'legacy', + 'canonical', + ]) + expect(existsSync(join(legacyDir, 'sync-ledger.json'))).toBe(false) + }) + + it('does not adopt a legacy XDG ledger when CODEBURN_CACHE_DIR is explicitly set', async () => { + const { readLedger, writeLedger } = await import('../src/sync/ledger.js') + const { existsSync, mkdirSync, writeFileSync } = await import('fs') + const explicitDir = join(tmpDir, 'explicit-cache-precedence') + const xdgDir = join(tmpDir, 'xdg-cache-precedence') + const legacyDir = join(xdgDir, 'codeburn') + + process.env.CODEBURN_CACHE_DIR = explicitDir + process.env.XDG_CACHE_HOME = xdgDir + mkdirSync(legacyDir, { recursive: true }) + writeFileSync(join(legacyDir, 'sync-ledger.json'), JSON.stringify([ + { key: 'legacy', ts: '2026-07-01T00:00:00Z' }, + ])) + + expect(readLedger()).toEqual([]) + writeLedger([{ key: 'explicit', ts: '2026-07-02T00:00:00Z' }]) + + expect(readLedger().map(entry => entry.key)).toEqual(['explicit']) + expect(existsSync(join(explicitDir, 'sync-ledger.json'))).toBe(true) + expect(existsSync(join(legacyDir, 'sync-ledger.json'))).toBe(true) + }) + + it('merges an XDG legacy ledger into an existing canonical ledger once', async () => { + const { readLedger } = await import('../src/sync/ledger.js') + const { existsSync, mkdirSync, writeFileSync } = await import('fs') + const canonicalDir = join(tmpDir, '.cache', 'codeburn') + const xdgDir = join(tmpDir, 'xdg-cache-merge') + const legacyDir = join(xdgDir, 'codeburn') + + delete process.env.CODEBURN_CACHE_DIR + process.env.XDG_CACHE_HOME = xdgDir + mkdirSync(canonicalDir, { recursive: true }) + mkdirSync(legacyDir, { recursive: true }) + writeFileSync(join(canonicalDir, 'sync-ledger.json'), JSON.stringify([ + { key: 'canonical', ts: '2026-07-02T00:00:00Z' }, + { key: 'duplicate', ts: '2026-07-03T00:00:00Z' }, + ])) + writeFileSync(join(legacyDir, 'sync-ledger.json'), JSON.stringify([ + { key: 'legacy', ts: '2026-07-01T00:00:00Z' }, + { key: 'duplicate', ts: '2025-01-01T00:00:00Z' }, + ])) + + expect(readLedger()).toEqual([ + { key: 'canonical', ts: '2026-07-02T00:00:00Z' }, + { key: 'duplicate', ts: '2026-07-03T00:00:00Z' }, + { key: 'legacy', ts: '2026-07-01T00:00:00Z' }, + ]) + expect(existsSync(join(legacyDir, 'sync-ledger.json'))).toBe(false) + // A later read is canonical-only and stable; XDG is no longer active. + expect(readLedger().map(entry => entry.key)).toEqual(['canonical', 'duplicate', 'legacy']) }) it('prefers non-empty CODEBURN_CACHE_DIR over XDG_CACHE_HOME', async () => { @@ -392,7 +543,7 @@ describe('ledger', () => { expect(existsSync(join(xdgDir, 'codeburn', 'sync-ledger.json'))).toBe(false) }) - it.each(['', ' '])('ignores empty CODEBURN_CACHE_DIR %j and falls back to XDG_CACHE_HOME', async explicit => { + it.each(['', ' '])('ignores empty CODEBURN_CACHE_DIR %j and writes to the canonical default', async explicit => { const { writeLedger } = await import('../src/sync/ledger.js') const { existsSync } = await import('fs') const { join } = await import('path') @@ -402,7 +553,8 @@ describe('ledger', () => { process.env.XDG_CACHE_HOME = xdgDir writeLedger([{ key: 'xdg-fallback', ts: '2026-07-01T00:00:00Z' }]) - expect(existsSync(join(xdgDir, 'codeburn', 'sync-ledger.json'))).toBe(true) + expect(existsSync(join(tmpDir, '.cache', 'codeburn', 'sync-ledger.json'))).toBe(true) + expect(existsSync(join(xdgDir, 'codeburn', 'sync-ledger.json'))).toBe(false) }) it.each(['', ' '])('ignores empty XDG_CACHE_HOME %j and uses the shared default', async xdg => {