import test from 'node:test'; import assert from 'node:assert/strict'; import { chmod, mkdtemp, rm, readFile, writeFile } from 'node:fs/promises'; import { spawn } from 'node:child_process'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { constants as fsConstants } from 'node:fs'; import { fileURLToPath } from 'node:url'; const root = fileURLToPath(new URL('..', import.meta.url)); const fixture = fileURLToPath(new URL('./fixtures/slow-hermes.mjs', import.meta.url)); const origin = 'http://127.0.0.1:4187'; const accessCode = 'queue-acceptance-code-2026'; async function waitReady(child) { const deadline = Date.now() + 10_000; while (Date.now() < deadline) { if (child.exitCode !== null) throw new Error(`server exited ${child.exitCode}`); try { const response = await fetch(`${origin}/api/healthz`); if (response.ok) return; } catch {} await new Promise(resolve => setTimeout(resolve, 50)); } throw new Error('server did not become ready'); } function post(path, body, { cookie = '', signal } = {}) { return fetch(`${origin}${path}`, { method: 'POST', headers: { 'content-type': 'application/json', origin, 'sec-fetch-site': 'same-origin', ...(cookie ? { cookie } : {}), }, body: JSON.stringify(body), signal, }).then( async response => ({ status: response.status, body: await response.json().catch(() => ({})) }), error => ({ failed: true, reason: String(error?.name || error) }), ); } test('bounded queue end-to-end: disconnect cleanup, overload fallback, urgent bypass, recovery', async t => { const workdir = await mkdtemp(join(tmpdir(), 'timmy-queue-accept-')); const releaseFile = join(workdir, 'hermes-release'); const pidLog = join(workdir, 'fixture-pids.log'); await chmod(fixture, fsConstants.S_IRWXU); const child = spawn(process.execPath, ['server.mjs'], { cwd: root, env: { ...process.env, PORT: '4187', TIMMY_AGENT_ENABLED: 'true', TIMMY_AGENT_ACCESS_TOKEN: accessCode, TIMMY_PUBLIC_ORIGIN: origin, TIMMY_AGENT_WORKDIR: workdir, TIMMY_HERMES_COMMAND: fixture, TIMMY_AGENT_MAX_CONCURRENT_TURNS: '1', TIMMY_AGENT_MAX_QUEUE_DEPTH: '2', }, stdio: ['ignore', 'pipe', 'pipe'], }); let serverLogs = ''; child.stderr.on('data', chunk => { serverLogs += String(chunk); }); t.after(async () => { child.kill('SIGTERM'); await rm(workdir, { recursive: true, force: true }).catch(() => {}); await rm(releaseFile, { force: true }).catch(() => {}); }); await waitReady(child); // Authenticate once and keep the cookie for the whole scenario. const rawResponse = await fetch(`${origin}/api/agent/unlock`, { method: 'POST', headers: { 'content-type': 'application/json', origin, 'sec-fetch-site': 'same-origin' }, body: JSON.stringify({ accessCode }), }); assert.equal(rawResponse.status, 200); const browserCookie = rawResponse.headers.get('set-cookie').split(';', 1)[0]; // Fill the slot: this HOLD runs until the fixture release file appears. const heldOne = post('/api/agent/chat', { message: 'HOLD one', ledger: [] }, { cookie: browserCookie }); await new Promise(resolve => setTimeout(resolve, 400)); // 1. Disconnect cleanup: this request waits in the queue. When its client // vanishes, Timmy must drop it and free the queued place. const disconnectController = new AbortController(); const doomed = post('/api/agent/chat', { message: 'HOLD doomed', ledger: [] }, { cookie: browserCookie, signal: disconnectController.signal }); await new Promise(resolve => setTimeout(resolve, 400)); disconnectController.abort(); const doomedOutcome = await doomed; assert.equal(doomedOutcome.failed, true, 'abandoned client saw its request cancelled'); // If the freed place was reclaimed, both follow-up HOLDs are admitted // (slot + queue of two minus one freed place). A sanitized 503 here would // mean the cancelled client did not release capacity. const heldTwo = post('/api/agent/chat', { message: 'HOLD two', ledger: [] }, { cookie: browserCookie }); const heldThree = post('/api/agent/chat', { message: 'HOLD three', ledger: [] }, { cookie: browserCookie }); await new Promise(resolve => setTimeout(resolve, 500)); // 2. Overload: everything is saturated now, so a further request must fail // fast with the stable manual-fallback copy and zero provider contact. const overloadResult = await post('/api/agent/chat', { message: 'overload probe', ledger: [] }, { cookie: browserCookie }); assert.equal(overloadResult.status, 503); assert.match(overloadResult.body.error, /busy|try again|manually/i); assert.doesNotMatch(overloadResult.body.error, /spawn|hermes|session|token|workdir|fixture/i); // 3. Urgent bypass while saturated: deterministic safety answer, zero calls. const urgentResult = await post('/api/agent/chat', { message: 'I have rectal bleeding', ledger: [] }, { cookie: browserCookie }); assert.equal(urgentResult.status, 200); assert.equal(urgentResult.body.safetyOverride, true); assert.match(urgentResult.body.reply, /medical help/i); // 4. Recovery: releasing the fixture lets every surviving request complete. await writeFile(releaseFile, 'go'); for (const [label, promise] of [['one', heldOne], ['two', heldTwo], ['three', heldThree]]) { const settled = await Promise.race([ promise, new Promise(resolve => setTimeout(() => resolve({ status: 'TEST-TIMEOUT' }), 12_000)), ]); assert.equal(settled.status, 200, `${label} completed after recovery`); assert.match(settled.body.reply, /bounded request/); assert.doesNotMatch(JSON.stringify(settled.body), /slow_fixture_2026|session_id/); } // 5. No orphan subprocesses: every fixture PID recorded at spawn is gone. await new Promise(resolve => setTimeout(resolve, 400)); const recordedPids = ((await readFile(pidLog, 'utf8').catch(() => '')) || '') .split('\n').map(Number).filter(Boolean); assert.equal(recordedPids.length, 3, `exactly three turns ran (saw ${recordedPids.length})`); for (const pid of [...new Set(recordedPids)]) { let alive = true; try { process.kill(pid, 0); } catch { alive = false; } assert.equal(alive, false, `fixture process ${pid} outlived its turn`); } assert.doesNotMatch(serverLogs, /access-code|secret|token/i); }, { timeout: 40_000 });