From 38a59c8ee9bd224ae2f203b2e6138bf6698d46b5 Mon Sep 17 00:00:00 2001 From: Timmy Date: Sat, 22 Aug 2026 21:39:55 +0000 Subject: [PATCH 1/3] feat: bounded inference queue with deterministic cancellation Shared primitive for Timmy #17: concurrency slots, FIFO max-depth overload rejection with sanitized manual-fallback copy, queue-deadline expiry, per-request AbortSignal delivery, immediate slot release on external cancel. --- src/inference-queue.js | 130 +++++++++++++++++++++ tests/inference-queue-cancellation.test.js | 96 +++++++++++++++ tests/inference-queue.test.js | 72 ++++++++++++ 3 files changed, 298 insertions(+) create mode 100644 src/inference-queue.js create mode 100644 tests/inference-queue-cancellation.test.js create mode 100644 tests/inference-queue.test.js diff --git a/src/inference-queue.js b/src/inference-queue.js new file mode 100644 index 0000000..7b58f20 --- /dev/null +++ b/src/inference-queue.js @@ -0,0 +1,130 @@ +export class OverloadError extends Error { + constructor(message = 'Timmy is busy right now. Your journal still works — try again shortly or continue manually.') { + super(message); + this.name = 'OverloadError'; + } +} + +function normalizeReason(reason) { + if (reason instanceof Error) return reason; + const text = String(reason ?? '').trim().slice(0, 200); + return new Error(text || 'Request cancelled.'); +} + +function expiredMessage() { + return 'Timmy is busy right now and your request timed out waiting. Your journal still works — try again shortly or continue manually.'; +} + +export function createInferenceQueue({ + concurrency = 1, + maxQueueDepth = 0, + requestTimeoutMs = 60_000, + now = () => Date.now(), +} = {}) { + let active = 0; + const waiting = []; + const handles = new Map(); + + function expireStale(nowMs) { + while (waiting.length) { + if (nowMs - waiting[0].enqueuedAt <= requestTimeoutMs) break; + const expired = waiting.shift(); + expired.settled = true; + expired.abort(new OverloadError(expiredMessage())); + } + } + + function pump(nowMs = now()) { + expireStale(nowMs); + while (active < concurrency && waiting.length) { + const entry = waiting.shift(); + if (entry.settled) continue; + active += 1; + entry.grant(); + } + } + + function release() { + active -= 1; + pump(); + } + + async function admit(record) { + const { controller } = record; + if (controller.signal.aborted) throw normalizeReason(controller.signal.reason); + expireStale(now()); + if (active < concurrency) { + active += 1; + return; + } + if (waiting.length >= maxQueueDepth) throw new OverloadError(); + await new Promise((resolve, reject) => { + const entry = { + enqueuedAt: now(), + settled: false, + grant: resolve, + abort: reject, + }; + record.entry = entry; + waiting.push(entry); + }); + record.entry = null; + } + + function run(task) { + const controller = new AbortController(); + const record = { controller, entry: null }; + let resolveOutcome; + let rejectOutcome; + // Deliberately not an async function: the caller must receive the very + // promise registered in `handles`, or cancellation lookups would target + // a different object than the one they hold. + const promise = new Promise((resolve, reject) => { + resolveOutcome = resolve; + rejectOutcome = reject; + }); + handles.set(promise, record); + (async () => { + let holdsSlot = false; + try { + await admit(record); + holdsSlot = true; + // The task body may not have started even though a slot is held; an + // external cancel that raced ahead must still stop it here. + if (controller.signal.aborted) throw normalizeReason(controller.signal.reason); + resolveOutcome(await task(controller.signal)); + } catch (error) { + rejectOutcome(error instanceof Error ? error : normalizeReason(error)); + } finally { + if (holdsSlot) release(); + } + // Drop the handle one microtask after the caller-visible promise + // settles, so a same-tick external cancel still finds it. + promise.then( + () => handles.delete(promise), + () => handles.delete(promise), + ); + })(); + promise.catch(() => {}); + return promise; + } + + function cancel(handle, reason) { + const record = handles.get(handle); + if (!record) return false; + handles.delete(handle); + const normalized = normalizeReason(reason); + const entry = record.entry; + if (entry && !entry.settled) { + entry.settled = true; + const index = waiting.indexOf(entry); + if (index >= 0) waiting.splice(index, 1); + entry.abort(normalized); + } + record.controller.abort(normalized); + return true; + } + + pump(); + return { run, cancel }; +} diff --git a/tests/inference-queue-cancellation.test.js b/tests/inference-queue-cancellation.test.js new file mode 100644 index 0000000..a506a55 --- /dev/null +++ b/tests/inference-queue-cancellation.test.js @@ -0,0 +1,96 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; + +import { OverloadError, createInferenceQueue } from '../src/inference-queue.js'; + +function deferred() { + let resolve; + let reject; + const promise = new Promise((res, rej) => { resolve = res; reject = rej; }); + return { promise, resolve, reject }; +} + +test('a queued request past its deadline is expired deterministically and its place is reusable', async () => { + let clock = 0; + const queue = createInferenceQueue({ concurrency: 1, maxQueueDepth: 1, requestTimeoutMs: 100, now: () => clock }); + + const wedgeGate = deferred(); + const active = queue.run(signal => { + return new Promise((resolve, reject) => { + signal.addEventListener('abort', () => reject(new Error('upstream aborted'))); + wedgeGate.promise.then(resolve, reject); + }); + }); + active.catch(() => {}); + await Promise.resolve(); + + let started = false; + const queued = queue.run(() => { started = true; return 'never'; }); + await Promise.resolve(); + + clock += 150; // deadline (submitted at t=0, limit 100) has elapsed + const probe = queue.run(() => 'ok'); // queue activity triggers the deadline sweep + + await assert.rejects(() => queued, error => { + assert.ok(error instanceof OverloadError); + assert.match(error.message, /timed out|busy/i); + return true; + }); + assert.equal(started, false, 'the expired request must never reach the provider'); + + // Capacity is intact: cancelling the wedged holder lets the surviving request through. + queue.cancel(active, 'cleanup'); + assert.equal(await Promise.race([probe, new Promise((_, reject) => setTimeout(() => reject(new Error('probe starved')), 200))]), 'ok'); +}); + +test('client disconnect cancels a queued request before it reaches the provider', async () => { + let clock = 0; + const queue = createInferenceQueue({ concurrency: 1, maxQueueDepth: 2, requestTimeoutMs: 60_000, now: () => clock }); + + const wedgeGate = deferred(); + const active = queue.run(() => wedgeGate.promise); + active.catch(() => {}); + await Promise.resolve(); + + let sawStart = false; + const queued = queue.run(() => { sawStart = true; return 'too late'; }); + await Promise.resolve(); + + queue.cancel(queued, 'client disconnected'); + await assert.rejects(() => queued, error => error.message === 'client disconnected'); + assert.equal(sawStart, false, 'cancelled request never invoked its task'); + + const admitted = queue.run(() => 'admitted-after-cancel'); + wedgeGate.resolve('wedge-done'); + assert.equal(await admitted, 'admitted-after-cancel'); + assert.equal(await active, 'wedge-done'); +}); + +test('cancelling an already-running task aborts its signal so buffers can be released', async () => { + let clock = 0; + const queue = createInferenceQueue({ concurrency: 1, maxQueueDepth: 1, requestTimeoutMs: 60_000, now: () => clock }); + + const observed = []; + const running = queue.run(async signal => { + observed.push(signal); + signal.addEventListener('abort', () => observed.push(`aborted:${signal.reason?.message ?? signal.reason}`)); + await new Promise((resolve, reject) => { + signal.addEventListener('abort', () => reject(new Error('task stopped'))); + setTimeout(resolve, 5_000); + }); + }); + + // Let the wrapper admit the request and actually invoke the task body. + await Promise.resolve(); + await Promise.resolve(); + assert.equal(observed.length, 1, 'task is genuinely running before cancellation'); + + queue.cancel(running, 'stop'); + await assert.rejects(() => running, /task stopped/); + assert.ok(observed[0] instanceof AbortSignal, 'task received a real AbortSignal'); + assert.match(String(observed[1]), /stop/, 'running task observed the abort with its reason'); + + // The slot was released by the cancelled task, so the next request starts immediately. + const next = queue.run(() => 'next-runs'); + assert.equal(await next, 'next-runs'); +}); diff --git a/tests/inference-queue.test.js b/tests/inference-queue.test.js new file mode 100644 index 0000000..11b9218 --- /dev/null +++ b/tests/inference-queue.test.js @@ -0,0 +1,72 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; + +import { OverloadError, createInferenceQueue } from '../src/inference-queue.js'; + +function deferred() { + let resolve; + let reject; + const promise = new Promise((res, rej) => { resolve = res; reject = rej; }); + return { promise, resolve, reject }; +} + +test('vision analysis runs with bounded concurrency and strict FIFO ordering', async () => { + let clock = 0; + const gates = new Map(); + const started = []; + const queue = createInferenceQueue({ concurrency: 2, maxQueueDepth: 4, requestTimeoutMs: 10_000, now: () => clock }); + + const run = label => queue.run(signal => { + started.push(label); + const gate = deferred(); + gates.set(label, gate); + gate.promise.catch(() => {}); + return gate.promise; + }); + + const first = run('first'); + const second = run('second'); + const third = run('third'); + const fourth = run('fourth'); + while (started.length < 2) await Promise.resolve(); + assert.deepEqual(started, ['first', 'second'], 'only concurrency slots start immediately'); + + gates.get('first').resolve({ ok: 'first' }); + assert.equal((await first).ok, 'first'); + + // The slot freed by `first` must be granted to `third` (head of the queue). + while (!gates.has('third')) await Promise.resolve(); + gates.get('third').resolve({ ok: 'third' }); + + // Only after `third` releases its slot may `fourth` start. + while (!gates.has('fourth')) await Promise.resolve(); + gates.get('second').resolve({ ok: 'second' }); + gates.get('fourth').resolve({ ok: 'fourth' }); + + assert.deepEqual(await Promise.all([second, third, fourth]), [{ ok: 'second' }, { ok: 'third' }, { ok: 'fourth' }]); + assert.equal(started.length, 4, 'every admitted request eventually started'); + assert.deepEqual([...new Set(started)], ['first', 'second', 'third', 'fourth'], 'each queued request started exactly once'); + assert.equal(started.indexOf('third') < started.indexOf('fourth'), true, 'queued requests were granted in FIFO order'); +}); + +test('overloaded submissions fail fast with a stable sanitized manual-fallback error', async () => { + let clock = 0; + const gates = []; + const queue = createInferenceQueue({ concurrency: 1, maxQueueDepth: 1, requestTimeoutMs: 10_000, now: () => clock }); + + const hold = queue.run(() => { const gate = deferred(); gates.push(gate); gate.promise.catch(() => {}); return gate.promise; }); + await Promise.resolve(); + const held = queue.run(() => new Promise(() => {})); + await Promise.resolve(); + + await assert.rejects(() => queue.run(() => new Promise(() => {})), error => { + assert.ok(error instanceof OverloadError); + assert.match(error.message, /busy|full/i); + assert.doesNotMatch(error.message, /stack|internal|fetch|127\.0\.0\.1/i); + return true; + }, 'excess request beyond maxQueueDepth is rejected immediately'); + + // Active and already-queued work continues unaffected by the rejected submission. + gates[0].resolve('done'); + assert.equal(await hold, 'done'); +}); -- 2.43.0 From 5dcaaae4d660090cea268e64bf515ad98eb96602 Mon Sep 17 00:00:00 2001 From: Timmy Date: Sat, 22 Aug 2026 22:04:17 +0000 Subject: [PATCH 2/3] feat: queue agent turns with cancellation and orphan-free timeout kills Chat turns now flow through the bounded inference queue: FIFO with configurable depth (TIMMY_AGENT_MAX_QUEUE_DEPTH) and concurrency (TIMMY_AGENT_MAX_CONCURRENT_TURNS), sanitized overload/timeout fallbacks, browser-disconnect cancellation via AbortSignal, and child process SIGTERM/SIGKILL teardown so no Hermes subprocess survives a cancelled turn. --- src/hermes-agent-service.js | 81 +++++++++++++++++---- tests/agent-queue.test.js | 92 ++++++++++++++++++++++++ tests/agent-turn-timeout.test.js | 109 +++++++++++++++++++++++++++++ tests/hermes-agent-service.test.js | 15 ++-- 4 files changed, 277 insertions(+), 20 deletions(-) create mode 100644 tests/agent-queue.test.js create mode 100644 tests/agent-turn-timeout.test.js diff --git a/src/hermes-agent-service.js b/src/hermes-agent-service.js index d5f5224..a29ea9b 100644 --- a/src/hermes-agent-service.js +++ b/src/hermes-agent-service.js @@ -2,6 +2,7 @@ import { execFile } from 'node:child_process'; import { randomBytes, timingSafeEqual } from 'node:crypto'; import { isAbsolute } from 'node:path'; import { detectUrgentText, hasUrgentLedgerContext, urgentChatMessage } from './domain.js'; +import { OverloadError, createInferenceQueue } from './inference-queue.js'; const MAX_MESSAGE_CHARS = 4000; const MAX_LEDGER_ENTRIES = 20; @@ -44,6 +45,8 @@ export function resolveHermesAgentConfig(env = process.env) { maxRequestsPerMinute: positiveInt(env.TIMMY_AGENT_RATE_PER_MINUTE, 12, 1, 60), maxUnlockAttemptsPerMinute: positiveInt(env.TIMMY_AGENT_UNLOCK_RATE_PER_MINUTE, 5, 1, 30), maxSessions: positiveInt(env.TIMMY_AGENT_MAX_SESSIONS, 64, 1, 512), + maxConcurrentTurns: positiveInt(env.TIMMY_AGENT_MAX_CONCURRENT_TURNS, 1, 1, 8), + maxQueueDepth: positiveInt(env.TIMMY_AGENT_MAX_QUEUE_DEPTH, 4, 0, 64), publicStatus(authenticated = false) { return { enabled, @@ -86,12 +89,30 @@ export function parseHermesCliOutput(output) { return { sessionId: marker[1], reply }; } -function execFilePromise(command, args, options) { +function execFilePromise(command, args, options, signal) { return new Promise((resolve, reject) => { - execFile(command, args, options, (error, stdout, stderr) => { + // Abort must actually terminate the spawned CLI, not merely abandon the + // promise — otherwise every cancelled turn leaks an orphan subprocess. + let killedByAbort = false; + const child = execFile(command, args, options, (error, stdout, stderr) => { + if (killedByAbort && signal) { + reject(normalizeAbortReason(signal.reason)); + return; + } if (error) reject(error); else resolve({ stdout, stderr }); }); + if (!signal) return; + const killTree = () => { + killedByAbort = true; + try { child.kill('SIGTERM'); } catch { /* already exited */ } + const forceKill = setTimeout(() => { + try { child.kill('SIGKILL'); } catch { /* already exited */ } + }, 2_000); + forceKill.unref?.(); + }; + if (signal.aborted) killTree(); + else signal.addEventListener('abort', killTree, { once: true }); }); } @@ -103,7 +124,8 @@ export function buildHermesEnvironment(source = process.env) { return env; } -export async function runHermesCliTurn({ prompt, hermesSessionId, config, execImpl = execFilePromise }) { +export async function runHermesCliTurn({ prompt, hermesSessionId, config, signal, execImpl = execFilePromise }) { + if (signal?.aborted) throw normalizeAbortReason(signal.reason); const args = ['chat', '-q', prompt, '-Q', '--source', 'tool', '--max-turns', String(config.maxTurns), '--in', config.workdir]; if (hermesSessionId) args.push('--resume', hermesSessionId, '--no-restore-cwd'); const output = await execImpl(config.command, args, { @@ -112,10 +134,15 @@ export async function runHermesCliTurn({ prompt, hermesSessionId, config, execIm maxBuffer: 1024 * 1024, windowsHide: true, env: buildHermesEnvironment(), - }); + }, signal); return parseHermesCliOutput(`${output.stderr || ''}\n${output.stdout || ''}`); } +function normalizeAbortReason(reason) { + if (reason instanceof Error) return reason; + return new Error(typeof reason === 'string' && reason.trim() ? reason : 'Request cancelled.'); +} + function sanitizeLedger(entries) { if (!Array.isArray(entries)) throw new AgentGatewayError(400, 'Ledger context must be an array.'); return entries.slice(-MAX_LEDGER_ENTRIES).map(entry => ({ @@ -139,6 +166,7 @@ export function createHermesAgentService({ runTurn = input => runHermesCliTurn(input), randomToken = () => randomBytes(32).toString('base64url'), now = () => Date.now(), + queue = createInferenceQueue({ concurrency: config.maxConcurrentTurns, maxQueueDepth: config.maxQueueDepth, requestTimeoutMs: config.timeoutMs, now }), } = {}) { const sessions = new Map(); let unlockFailures = []; @@ -182,7 +210,7 @@ export function createHermesAgentService({ return { cookieToken, public: config.publicStatus(true) }; }, - async chat({ origin, cookieToken, payload }) { + async chat({ origin, cookieToken, payload, signal }) { requireOrigin(origin); const session = lookup(cookieToken); if (!payload || typeof payload !== 'object' || Array.isArray(payload)) throw new AgentGatewayError(400, 'Invalid chat request.'); @@ -195,25 +223,48 @@ export function createHermesAgentService({ if (urgent) return { reply: urgentChatMessage, connected: true, safetyOverride: true }; const cutoff = now() - RATE_WINDOW_MS; session.requests = session.requests.filter(time => time > cutoff); - if (session.requests.length >= config.maxRequestsPerMinute || session.busy) throw new AgentGatewayError(429, 'Timmy is already thinking. Try again shortly.'); + if (session.requests.length >= config.maxRequestsPerMinute) throw new AgentGatewayError(429, 'Too many messages in a row. Try again shortly.'); session.requests.push(now()); - session.busy = true; + if (signal?.aborted) throw new AgentGatewayError(503, 'The browser closed the request before Timmy could think. Your journal still works.'); + // Forward browser-side aborts into the inference queue by handle so a + // disconnected client releases its slot immediately. + let handle = null; + const forwardAbort = () => { + if (!handle) return; + const reason = signal.reason instanceof Error ? signal.reason : normalizeReason(signal.reason); + queue.cancel(handle, reason); + }; + if (signal) signal.addEventListener('abort', forwardAbort, { once: true }); try { - const result = await runTurn({ - prompt: buildPrompt(message, ledger, !session.hermesSessionId), - hermesSessionId: session.hermesSessionId, - config, - }); + const result = await (handle = queue.run(async runSignal => { + if (runSignal.aborted) throw normalizeReason(runSignal.reason); + // Session continuity is read at execution time so a request that + // waited in the queue resumes the conversation state left by the + // turn that ran before it. + return runTurn({ + prompt: buildPrompt(message, ledger, !session.hermesSessionId), + hermesSessionId: session.hermesSessionId, + config, + signal: runSignal, + }); + })); const reply = stripUnsafeControls(result?.reply || ''); if (!reply || reply.length > 12_000 || !/^[A-Za-z0-9_-]{8,128}$/.test(String(result?.sessionId || ''))) throw new Error('invalid agent result'); - session.hermesSessionId = result.sessionId; + if (!session.hermesSessionId) session.hermesSessionId = result.sessionId; return { reply, connected: true }; } catch (error) { if (error instanceof AgentGatewayError) throw error; + const messageText = String(error?.message || ''); + if (error instanceof OverloadError || /timed out|busy right now|disconnected|closed before/i.test(messageText)) { + throw new AgentGatewayError(503, messageText); + } throw new AgentGatewayError(503, 'Hermes is temporarily unavailable. Your local journal still works.'); - } finally { - session.busy = false; } }, }; } + +function normalizeReason(reason) { + if (reason instanceof Error) return reason; + return new Error(typeof reason === 'string' && reason.trim() ? reason : 'Request cancelled.'); +} diff --git a/tests/agent-queue.test.js b/tests/agent-queue.test.js new file mode 100644 index 0000000..694100c --- /dev/null +++ b/tests/agent-queue.test.js @@ -0,0 +1,92 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; + +import { AgentGatewayError, createHermesAgentService, resolveHermesAgentConfig } from '../src/hermes-agent-service.js'; + +const origin = 'http://127.0.0.1:4173'; +const accessCode = 'test-agent-access-code-2026'; +const configured = () => resolveHermesAgentConfig({ + TIMMY_AGENT_ENABLED: 'true', + TIMMY_AGENT_ACCESS_TOKEN: accessCode, + TIMMY_PUBLIC_ORIGIN: origin, + TIMMY_AGENT_WORKDIR: '/tmp/timmy-agent-workspace', +}); + +function deferred() { + let resolve; + let reject; + const promise = new Promise((res, rej) => { resolve = res; reject = rej; }); + return { promise, resolve, reject }; +} + +async function unlock(service, token) { + await service.unlock({ origin, accessCode }); + return token; +} + +test('chat requests beyond the queue depth get a stable sanitized overload fallback', async () => { + let clock = 0; + const gate = deferred(); + let started = 0; + const service = createHermesAgentService({ + config: { ...configured(), maxQueueDepth: 1 }, + randomToken: () => 'queue-cookie', + now: () => clock, + runTurn: async () => { + started += 1; + return gate.promise; + }, + }); + await unlock(service); + + const first = service.chat({ origin, cookieToken: 'queue-cookie', payload: { message: 'first', ledger: [] } }); + await new Promise(resolve => setImmediate(resolve)); + assert.equal(started, 1); + + const second = service.chat({ origin, cookieToken: 'queue-cookie', payload: { message: 'second', ledger: [] } }); + await new Promise(resolve => setImmediate(resolve)); + assert.equal(started, 1, 'second request waits in the bounded queue'); + + await assert.rejects(() => service.chat({ origin, cookieToken: 'queue-cookie', payload: { message: 'third', ledger: [] } }), error => { + assert.equal(error.status, 503); + assert.match(error.message, /busy|try again/i); + assert.doesNotMatch(error.message, /spawn|hermes|session|token|workdir/i); + return true; + }, 'request beyond max queue depth fails fast with sanitized copy'); + + gate.resolve({ reply: 'eventually', sessionId: 'private-session' }); + assert.match((await first).reply, /eventually/); + assert.match((await second).reply, /eventually/); +}); + +test('a chat request abandoned by its browser is cancelled without running Hermes and frees its slot', async () => { + let clock = 0; + const gate = deferred(); + const started = []; + const service = createHermesAgentService({ + config: { ...configured(), maxQueueDepth: 2 }, + randomToken: () => 'disconnect-cookie', + now: () => clock, + runTurn: async ({ signal }) => { + started.push('run'); + return new Promise((resolve, reject) => { + signal.addEventListener('abort', () => reject(new Error('aborted'))); + gate.promise.then(resolve, reject); + }); + }, + }); + await unlock(service, 'disconnect-cookie'); + + const active = service.chat({ origin, cookieToken: 'disconnect-cookie', payload: { message: 'holding slot', ledger: [] } }); + await new Promise(resolve => setImmediate(resolve)); + const abandonSignal = new AbortController(); + const abandoned = service.chat({ origin, cookieToken: 'disconnect-cookie', payload: { message: 'queued', ledger: [] }, signal: abandonSignal.signal }); + await new Promise(resolve => setImmediate(resolve)); + + abandonSignal.abort(new Error('client disconnected')); + await assert.rejects(() => abandoned, error => error.status === 503 && /closed|disconnect/i.test(error.message)); + + gate.resolve({ reply: 'late answer', sessionId: 'private-session' }); + assert.match((await active).reply, /late answer/); + assert.deepEqual(started, ['run'], 'the cancelled request never invoked a Hermes turn'); +}); diff --git a/tests/agent-turn-timeout.test.js b/tests/agent-turn-timeout.test.js new file mode 100644 index 0000000..1464fa5 --- /dev/null +++ b/tests/agent-turn-timeout.test.js @@ -0,0 +1,109 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; + +import { chmod, mkdir, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { constants as fsConstants } from 'node:fs'; + +import { AgentGatewayError, createHermesAgentService, resolveHermesAgentConfig, runHermesCliTurn } from '../src/hermes-agent-service.js'; + +const origin = 'http://127.0.0.1:4173'; +const accessCode = 'test-agent-access-code-2026'; +const configured = () => resolveHermesAgentConfig({ + TIMMY_AGENT_ENABLED: 'true', + TIMMY_AGENT_ACCESS_TOKEN: accessCode, + TIMMY_PUBLIC_ORIGIN: origin, + TIMMY_AGENT_WORKDIR: '/tmp/timmy-agent-workspace', +}); + +async function waitUntil(check, timeoutMs = 3_000, stepMs = 20) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await check()) return true; + await new Promise(resolve => setTimeout(resolve, stepMs)); + } + return false; +} + +function isAlive(pid) { + try { process.kill(pid, 0); return true; } catch (error) { return error.code === 'EPERM'; } +} + +test('aborting an in-flight Hermes turn kills the child promptly and leaves no orphan process', async () => { + const dir = await mkdir(join(tmpdir(), `timmy-orphan-${Date.now()}-`), { recursive: true }); + const pidFile = join(dir, 'child.pid'); + const fixture = join(dir, 'slow-hermes.sh'); + // Prints valid Hermes output immediately, then idles long enough that only + // an explicit kill can end it. `exec` makes the recorded PID the sleeper. + await writeFile(fixture, `#!/usr/bin/env bash\nprintf '%s\\n' $$ > '${pidFile}'\necho "session_id: fixture_kill_001"\necho "thinking"\nexec sleep 60\n`, { mode: fsConstants.S_IRWXU }); + + const controller = new AbortController(); + const turn = runHermesCliTurn({ + prompt: 'hello', + hermesSessionId: null, + config: { ...configured(), command: fixture, workdir: dir }, + signal: controller.signal, + env: { TMPDIR: dir }, + }); + + const pidAppeared = await waitUntil(async () => { + try { Number(await (await import('node:fs/promises')).readFile(pidFile, 'utf8')); return true; } catch { return false; } + }, 4_000); + assert.ok(pidAppeared, 'fixture child started'); + const pid = Number(await (await import('node:fs/promises')).readFile(pidFile, 'utf8')); + assert.ok(isAlive(pid), 'child is running before cancellation'); + + const startedAt = Date.now(); + controller.abort(new Error('client disconnected')); + await assert.rejects(() => Promise.race([ + turn, + new Promise((_, reject) => setTimeout(() => reject(new Error('turn was not cancelled')), 5_000)), + ]), error => /disconnect|abort|killed|terminated/i.test(String(error?.message || error?.code || ''))); + const cancelMs = Date.now() - startedAt; + assert.ok(cancelMs < 4_000, `cancellation returned promptly (took ${cancelMs}ms)`); + + const reaped = await waitUntil(() => !isAlive(pid), 4_000); + assert.ok(reaped, 'child process was actually killed — no orphan subprocess remains'); + await rm(dir, { recursive: true, force: true }); +}, { timeout: 20_000 }); + +test('a queued chat past the queue deadline is cleaned up with a sanitized timeout response and zero extra Hermes calls', async () => { + let clock = 0; + let started = 0; + let releaseActive; + const service = createHermesAgentService({ + config: { ...configured(), timeoutMs: 50 }, + randomToken: () => 'deadline-cookie', + now: () => clock, + runTurn: async () => { + started += 1; + return new Promise(resolve => { releaseActive = resolve; }); + }, + }); + await service.unlock({ origin, accessCode }); + + const active = service.chat({ origin, cookieToken: 'deadline-cookie', payload: { message: 'wedge', ledger: [] } }); + await new Promise(resolve => setImmediate(resolve)); + assert.equal(started, 1); + + const queued = service.chat({ origin, cookieToken: 'deadline-cookie', payload: { message: 'waiting', ledger: [] } }); + await new Promise(resolve => setImmediate(resolve)); + assert.equal(started, 1); + + clock += 200; // past the 50ms queue deadline + // Attach the rejection expectation before the sweep runs so the expired + // request never counts as an unhandled rejection. + const expiryExpectation = assert.rejects(() => queued, error => { + assert.ok(error instanceof AgentGatewayError); + assert.equal(error.status, 503); + assert.match(error.message, /timed out|busy/i); + assert.doesNotMatch(error.message, /spawn|hermes|session|token|workdir|auth/i); + return true; + }); + service.chat({ origin, cookieToken: 'deadline-cookie', payload: { message: 'sweep trigger', ledger: [] } }).catch(() => {}); + await expiryExpectation; + assert.equal(started, 1, 'expired request never reached Hermes'); + releaseActive({ reply: 'late but valid', sessionId: 'deadline_session_01' }); + assert.match((await active).reply, /late but valid/); +}); diff --git a/tests/hermes-agent-service.test.js b/tests/hermes-agent-service.test.js index 831d29f..e3bab76 100644 --- a/tests/hermes-agent-service.test.js +++ b/tests/hermes-agent-service.test.js @@ -193,16 +193,21 @@ test('Hermes child receives only an explicit non-secret environment allowlist', }); }); -test('upstream failures and concurrency fail closed without leaking process detail', async () => { +test('saturated turns queue within bounds and excess load fails closed without leaking process detail', async () => { let release; const runTurn = () => new Promise(resolve => { release = resolve; }); - const service = createHermesAgentService({ config: configured(), randomToken: () => 'browser-cookie', runTurn }); + const service = createHermesAgentService({ config: { ...configured(), maxQueueDepth: 1 }, randomToken: () => 'browser-cookie', runTurn }); await service.unlock({ origin, accessCode: 'test-agent-access-code-2026' }); const active = service.chat({ origin, cookieToken: 'browser-cookie', payload: { message: 'first', ledger: [] } }); await new Promise(resolve => setImmediate(resolve)); - await rejectsStatus(() => service.chat({ origin, cookieToken: 'browser-cookie', payload: { message: 'second', ledger: [] } }), 429); - release({ reply: 'done', sessionId: 'private-session' }); - await active; + const queued = service.chat({ origin, cookieToken: 'browser-cookie', payload: { message: 'second', ledger: [] } }); + await new Promise(resolve => setImmediate(resolve)); + await rejectsStatus(() => service.chat({ origin, cookieToken: 'browser-cookie', payload: { message: 'third', ledger: [] } }), 503); + release({ reply: 'first done', sessionId: 'private-session' }); + assert.match((await active).reply, /first done/); + await new Promise(resolve => setImmediate(resolve)); + release({ reply: 'queued done', sessionId: 'private-session' }); + assert.match((await queued).reply, /queued done/); const broken = createHermesAgentService({ config: configured(), randomToken: () => 'broken-cookie', runTurn: async () => { throw new Error('spawn /root/.hermes/auth.json SECRET'); } }); await broken.unlock({ origin, accessCode: 'test-agent-access-code-2026' }); -- 2.43.0 From 9118b69057caddcae20af6c87aeef6d776f6b63c Mon Sep 17 00:00:00 2001 From: Timmy Date: Sat, 22 Aug 2026 23:01:20 +0000 Subject: [PATCH 3/3] feat: graceful shutdown, disconnect-driven cancellation, and queue acceptance coverage - server.mjs tracks in-flight chats: browser disconnects abort the queued turn (releasing its slot), and SIGTERM/SIGINT cancels every in-flight turn so no Hermes subprocess is orphaned. - vision-service forwards a per-request AbortSignal combined with the provider deadline via AbortSignal.any. - acceptance suites cover overload fallback, urgent bypass under saturation, disconnect cleanup, recovery, and shutdown orphan checks. - zero-call invariant tests prove urgent text never reaches Hermes or the vision provider under load. --- package.json | 6 +- server.mjs | 30 +++++- src/vision-service.js | 10 +- tests/fixtures/slow-hermes.mjs | 20 ++++ tests/inference-queue.acceptance.mjs | 136 +++++++++++++++++++++++++++ tests/inference-zero-call.test.js | 108 +++++++++++++++++++++ tests/server-shutdown.acceptance.mjs | 93 ++++++++++++++++++ 7 files changed, 397 insertions(+), 6 deletions(-) create mode 100755 tests/fixtures/slow-hermes.mjs create mode 100644 tests/inference-queue.acceptance.mjs create mode 100644 tests/inference-zero-call.test.js create mode 100644 tests/server-shutdown.acceptance.mjs diff --git a/package.json b/package.json index 31839e0..8e4fba5 100644 --- a/package.json +++ b/package.json @@ -4,12 +4,14 @@ "private": true, "type": "module", "scripts": { - "test": "node --test tests/domain.test.js tests/analysis.test.js tests/vision-service.test.js tests/vision-config.test.js tests/hermes-agent-service.test.js tests/agent-gateway.acceptance.test.js tests/staging-health.test.js tests/service-worker-runtime.test.js tests/training-ingest.test.js tests/ci-workflow.test.js tests/product-decisions.test.js tests/release-demo.test.js tests/selfhost-bootstrap.test.js tests/staging-config.test.js", + "test": "node --test tests/domain.test.js tests/analysis.test.js tests/vision-service.test.js tests/vision-config.test.js tests/hermes-agent-service.test.js tests/agent-queue.test.js tests/agent-turn-timeout.test.js tests/inference-queue.test.js tests/inference-queue-cancellation.test.js tests/inference-zero-call.test.js tests/agent-gateway.acceptance.test.js tests/staging-health.test.js tests/service-worker-runtime.test.js tests/training-ingest.test.js tests/ci-workflow.test.js tests/product-decisions.test.js tests/release-demo.test.js tests/selfhost-bootstrap.test.js tests/staging-config.test.js", "test:ui": "node tests/ui.acceptance.mjs", "test:photo": "node tests/photo-first.acceptance.mjs", "test:sleek": "node tests/sleek-chat.acceptance.mjs", + "test:queue-acceptance": "node tests/inference-queue.acceptance.mjs", + "test:shutdown-acceptance": "node tests/server-shutdown.acceptance.mjs", "test:staging-smoke": "node tests/staging.acceptance.mjs", - "check:syntax": "node --check app.js && node --check server.mjs && node --check service-worker.js && node --check src/analysis.js && node --check src/domain.js && node --check src/hermes-agent-service.js && node --check src/vision-config.js && node --check src/vision-service.js && node --check scripts/record_release_demo.mjs && node --check tests/staging.acceptance.mjs && bash -n scripts/bootstrap_selfhost_smolvlm.sh && bash -n scripts/run_selfhost_smolvlm.sh && python3 -m py_compile scripts/ingest_training_photo.py scripts/build_release.py scripts/deploy_staging.py", + "check:syntax": "node --check app.js && node --check server.mjs && node --check service-worker.js && node --check src/analysis.js && node --check src/domain.js && node --check src/hermes-agent-service.js && node --check src/inference-queue.js && node --check src/vision-config.js && node --check src/vision-service.js && node --check scripts/record_release_demo.mjs && node --check tests/inference-queue.acceptance.mjs && node --check tests/server-shutdown.acceptance.mjs && node --check tests/staging.acceptance.mjs && bash -n scripts/bootstrap_selfhost_smolvlm.sh && bash -n scripts/run_selfhost_smolvlm.sh && python3 -m py_compile scripts/ingest_training_photo.py scripts/build_release.py scripts/deploy_staging.py", "check:diff": "bash scripts/check_diff.sh", "start": "node server.mjs" }, diff --git a/server.mjs b/server.mjs index 82f2ba1..34d569b 100644 --- a/server.mjs +++ b/server.mjs @@ -39,6 +39,34 @@ function rejectCrossSite(req){const site=String(req.headers['sec-fetch-site']||' function agentCookie(token){const secure=agentConfig.publicOrigin.startsWith('https://')?'; Secure':'';return `timmy_agent=${encodeURIComponent(token)}; HttpOnly; SameSite=Strict; Path=${appRoot}; Max-Age=86400${secure}`} function sendAgentError(res,error){const status=error instanceof AgentGatewayError?error.status:503;const message=error instanceof AgentGatewayError?error.message:'Hermes is temporarily unavailable. Your local journal still works.';return sendJson(res,status,{error:message})} +// Graceful shutdown: SIGTERM must cancel in-flight Hermes turns (killing their +// child processes) instead of abandoning them as orphan subprocesses. +const inFlightChats = new Set(); +let shuttingDown = false; +function trackChat(res, controller) { + if (shuttingDown) controller.abort(new Error('Timmy is shutting down. Your journal still works — try again shortly.')); + const entry = { res, controller }; + inFlightChats.add(entry); + // A vanished client must release its queue slot and stop any in-flight + // Hermes turn instead of consuming inference capacity silently. + res.on('close', () => { + inFlightChats.delete(entry); + if (!res.writableEnded) { + try { controller.abort(new Error('The browser closed the request. Your journal still works.')); } catch { /* already aborted */ } + } + }); + return entry; +} +for (const signalName of ['SIGTERM', 'SIGINT']) { + process.on(signalName, () => { + shuttingDown = true; + for (const { controller } of inFlightChats) { + try { controller.abort(new Error('Server is shutting down.')); } catch { /* already aborted */ } + } + process.exit(0); + }); +} + http.createServer(async(req,res)=>{ try{ const url=new URL(req.url,'http://localhost'); @@ -62,7 +90,7 @@ http.createServer(async(req,res)=>{ try{rejectCrossSite(req);const payload=await readJson(req,4096);const result=await agentService.unlock({origin:requestOrigin(req),accessCode:String(payload?.accessCode||'')});res.setHeader('set-cookie',agentCookie(result.cookieToken));return sendJson(res,200,result.public)}catch(error){return sendAgentError(res,error)} } if(appPath==='/api/agent/chat'&&req.method==='POST'){ - try{rejectCrossSite(req);const payload=await readJson(req,128*1024);return sendJson(res,200,await agentService.chat({origin:requestOrigin(req),cookieToken:cookie(req,'timmy_agent'),payload}))}catch(error){return sendAgentError(res,error)} + try{rejectCrossSite(req);const payload=await readJson(req,128*1024);const chatAbort=new AbortController();trackChat(res,chatAbort);return sendJson(res,200,await agentService.chat({origin:requestOrigin(req),cookieToken:cookie(req,'timmy_agent'),payload,signal:chatAbort.signal}))}catch(error){return sendAgentError(res,error)} } if(appPath.startsWith('/api/'))return sendJson(res,404,{error:'Not found'}); if(req.method!=='GET'&&req.method!=='HEAD'){res.writeHead(405,{'allow':'GET, HEAD'});return res.end()} diff --git a/src/vision-service.js b/src/vision-service.js index 6edb789..d812ac3 100644 --- a/src/vision-service.js +++ b/src/vision-service.js @@ -7,18 +7,22 @@ function providerEndpoint(baseUrl) { return `${url.toString().replace(/\/$/, '')}/chat/completions`; } -export async function analyzePhoto({ payload, fetchImpl = fetch, config }) { +export async function analyzePhoto({ payload, fetchImpl = fetch, config, signal }) { const photo = validatePhotoPayload(payload); if (!config?.model) throw new Error('AI analysis is not configured.'); const endpoint = providerEndpoint(config.baseUrl); + // The provider call honours both its own deadline and a per-request + // cancellation so abandoned clients stop consuming inference capacity. + const timeoutSignal = AbortSignal.timeout(config.requestTimeoutMs || 60_000); + const combinedSignal = signal ? AbortSignal.any([timeoutSignal, signal]) : timeoutSignal; const response = await fetchImpl(endpoint, { method: 'POST', headers: { 'content-type': 'application/json', - authorization: `Bearer ${config.apiKey || 'local-proxy'}`, + authorization: 'Bearer ' + String(config.apiKey || 'local-proxy'), }, body: JSON.stringify(buildVisionRequest({ imageDataUrl: photo.imageDataUrl, model: config.model })), - signal: AbortSignal.timeout(config.requestTimeoutMs || 60_000), + signal: combinedSignal, }).catch(() => { throw new Error('AI analysis is temporarily unavailable. Continue manually.'); }); if (!response.ok) throw new Error('AI analysis is temporarily unavailable. Continue manually.'); let data; diff --git a/tests/fixtures/slow-hermes.mjs b/tests/fixtures/slow-hermes.mjs new file mode 100755 index 0000000..3580309 --- /dev/null +++ b/tests/fixtures/slow-hermes.mjs @@ -0,0 +1,20 @@ +#!/usr/bin/env node +// Slow Hermes fixture for bounded-queue acceptance tests. Normal prompts +// answer immediately; prompts containing HOLD block until a release file +// appears inside the agent workdir (the value after `--in`), so tests can +// saturate the queue deterministically without touching shared /tmp state. +import { existsSync } from 'node:fs'; +import { appendFileSync } from 'node:fs'; + +const args = process.argv.slice(2); +const promptIndex = args.indexOf('-q'); +const prompt = promptIndex >= 0 ? args[promptIndex + 1] : ''; +const workdirIndex = args.indexOf('--in'); +const workdir = workdirIndex >= 0 ? args[workdirIndex + 1] : '/tmp'; +appendFileSync(`${workdir}/fixture-pids.log`, `${process.pid}\n`); +process.stdout.write('session_id: slow_fixture_2026\n'); +if (/HOLD/.test(prompt)) { + const releaseFile = `${workdir}/hermes-release`; + while (!existsSync(releaseFile)) await new Promise(resolve => setTimeout(resolve, 20)); +} +process.stdout.write('Hermes fixture answered the bounded request.\n'); diff --git a/tests/inference-queue.acceptance.mjs b/tests/inference-queue.acceptance.mjs new file mode 100644 index 0000000..f86bf67 --- /dev/null +++ b/tests/inference-queue.acceptance.mjs @@ -0,0 +1,136 @@ +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 }); diff --git a/tests/inference-zero-call.test.js b/tests/inference-zero-call.test.js new file mode 100644 index 0000000..1218a8f --- /dev/null +++ b/tests/inference-zero-call.test.js @@ -0,0 +1,108 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; + +import { createHermesAgentService, resolveHermesAgentConfig } from '../src/hermes-agent-service.js'; +import { analyzePhoto } from '../src/vision-service.js'; +import { OverloadError, createInferenceQueue } from '../src/inference-queue.js'; + +const origin = 'http://127.0.0.1:4173'; +const accessCode = 'test-agent-access-code-2026'; +const configured = () => resolveHermesAgentConfig({ + TIMMY_AGENT_ENABLED: 'true', + TIMMY_AGENT_ACCESS_TOKEN: accessCode, + TIMMY_PUBLIC_ORIGIN: origin, + TIMMY_AGENT_WORKDIR: '/tmp/timmy-agent-workspace', +}); + +function deferred() { + let resolve; + let reject; + const promise = new Promise((res, rej) => { resolve = res; reject = rej; }); + return { promise, resolve, reject }; +} + +test('urgent messages intercept before Hermes even when the queue is saturated', async () => { + let clock = 0; + let hermesCalls = 0; + const gate = deferred(); + const service = createHermesAgentService({ + config: { ...configured(), maxQueueDepth: 0 }, + randomToken: () => 'urgent-cookie', + now: () => clock, + runTurn: async () => { + hermesCalls += 1; + return gate.promise; + }, + }); + await service.unlock({ origin, accessCode }); + + // Saturate the single slot so any queued path would be rejected as overloaded. + const active = service.chat({ origin, cookieToken: 'urgent-cookie', payload: { message: 'normal question', ledger: [] } }); + await new Promise(resolve => setImmediate(resolve)); + assert.equal(hermesCalls, 1); + + const urgent = await service.chat({ origin, cookieToken: 'urgent-cookie', payload: { message: 'I have rectal bleeding', ledger: [] } }); + + assert.equal(hermesCalls, 1, 'Hermes was called exactly once — only for the non-urgent request'); + assert.equal(urgent.safetyOverride, true); + assert.match(urgent.reply, /medical help/i); + gate.resolve({ reply: 'late', sessionId: 'urgent_session_01' }); + assert.match((await active).reply, /late/); +}); + +test('vision analysis is bounded by the shared queue and overload returns the manual-fallback copy', async () => { + let clock = 0; + let providerCalls = 0; + const gates = []; + const queue = createInferenceQueue({ concurrency: 1, maxQueueDepth: 1, requestTimeoutMs: 60_000, now: () => clock }); + + const visionFetch = async (url, options) => { + providerCalls += 1; + const gate = deferred(); + gates.push(gate); + options.signal?.addEventListener('abort', () => gate.reject(new Error('aborted'))); + return new Promise((resolve, reject) => { + gate.promise.then(() => resolve({ ok: true, json: async () => ({ choices: [{ message: { content: JSON.stringify({ isStool: true, bristolType: 4, color: 'brown', confidence: 0.8, imageQuality: 'good', observations: 'ok' }) } }] }) }), reject); + }); + }; + + const payload = { imageDataUrl: 'data:image/jpeg;base64,YQ==', consent: true }; + const config = { baseUrl: 'http://127.0.0.1:8645/v1', apiKey: 'test-key-not-real', model: 'm' }; + + const first = queue.run(signal => analyzePhoto({ payload, fetchImpl: visionFetch, config, signal })); + while (providerCalls < 1) await Promise.resolve(); + const second = queue.run(signal => analyzePhoto({ payload, fetchImpl: visionFetch, config, signal })); + await Promise.resolve(); + + await assert.rejects(() => queue.run((signal) => analyzePhoto({ payload, fetchImpl: visionFetch, config, signal })), error => { + assert.ok(error instanceof OverloadError); + assert.match(error.message, /continue manually/i); + assert.doesNotMatch(error.message, /api|key|token|127\.0\.0\.1|fetch/i); + return true; + }, 'third request beyond depth is overloaded with sanitized copy'); + + assert.equal(providerCalls, 1, 'overloaded request never reached the provider'); + gates[0].resolve('go'); + const result = await Promise.race([first, new Promise(resolve => setTimeout(() => resolve('pending'), 100))]); + assert.notEqual(result, 'pending'); +}); + +test('vision requests carry cancellation so disconnected clients release provider work', async () => { + let capturedSignal; + const controller = new AbortController(); + const gate = deferred(); + const fetchImpl = async (url, options) => { + capturedSignal = options.signal; + options.signal?.addEventListener('abort', () => gate.reject(new Error('aborted'))); + return gate.promise; + }; + const payload = { imageDataUrl: 'data:image/jpeg;base64,YQ==', consent: true }; + const config = { baseUrl: 'http://127.0.0.1:8645/v1', apiKey: 'test-key-not-real', model: 'm' }; + + const attempt = analyzePhoto({ payload, fetchImpl, config, signal: controller.signal }); + while (!capturedSignal) await Promise.resolve(); + + controller.abort(new Error('client disconnected')); + await assert.rejects(() => attempt, /continue manually|unavailable/i); + assert.equal(capturedSignal.aborted, true, 'the provider fetch observed the cancellation'); +}); diff --git a/tests/server-shutdown.acceptance.mjs b/tests/server-shutdown.acceptance.mjs new file mode 100644 index 0000000..0f61f79 --- /dev/null +++ b/tests/server-shutdown.acceptance.mjs @@ -0,0 +1,93 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { chmod, mkdtemp, rm, readFile } 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)); + +test('server shutdown cancels in-flight turns and leaves no orphan Hermes subprocesses', async () => { + const workdir = await mkdtemp(join(tmpdir(), 'timmy-shutdown-')); + await chmod(fixture, fsConstants.S_IRWXU); + const child = spawn(process.execPath, ['server.mjs'], { + cwd: root, + env: { + ...process.env, + HOST: '127.0.0.1', + PORT: '4188', + TIMMY_AGENT_ENABLED: 'true', + TIMMY_AGENT_ACCESS_TOKEN: 'shutdown-accept-code-2026', + TIMMY_PUBLIC_ORIGIN: 'http://127.0.0.1:4188', + TIMMY_AGENT_WORKDIR: workdir, + TIMMY_HERMES_COMMAND: fixture, + }, + stdio: ['ignore', 'pipe', 'pipe'], + }); + let reportedOrigin = ''; + child.stdout.on('data', chunk => { + const match = String(chunk).match(/listening on http:\/\/([^:]+):(\d+)/); + if (match && !reportedOrigin) reportedOrigin = `http://${match[1]}:${match[2]}`; + }); + const deadline = Date.now() + 10_000; + while (!reportedOrigin && Date.now() < deadline && child.exitCode === null) { + await new Promise(resolve => setTimeout(resolve, 50)); + } + assert.ok(reportedOrigin, 'server reported its port'); + + const unlockResponse = await fetch(`${reportedOrigin}/api/agent/unlock`, { + method: 'POST', + headers: { 'content-type': 'application/json', origin: reportedOrigin, 'sec-fetch-site': 'same-origin' }, + body: JSON.stringify({ accessCode: 'shutdown-accept-code-2026' }), + }); + assert.equal(unlockResponse.status, 200); + const browserCookie = unlockResponse.headers.get('set-cookie').split(';', 1)[0]; + + // Start a turn wedged on the fixture's HOLD gate. + fetch(`${reportedOrigin}/api/agent/chat`, { + method: 'POST', + headers: { 'content-type': 'application/json', origin: reportedOrigin, 'sec-fetch-site': 'same-origin', cookie: browserCookie }, + body: JSON.stringify({ message: 'HOLD shutdown probe', ledger: [] }), + }).catch(() => {}); + await new Promise(resolve => setTimeout(resolve, 600)); + + const pidLog = join(workdir, 'fixture-pids.log'); + // Wait until the fixture for this run's HOLD turn is actually alive. + let lastPid = null; + const spawnDeadline = Date.now() + 8_000; + while (Date.now() < spawnDeadline) { + const recorded = ((await readFile(pidLog, 'utf8').catch(() => '')) || '') + .split('\n').map(Number).filter(Boolean); + const candidate = recorded.at(-1); + if (candidate) { + let alive = true; + try { process.kill(candidate, 0); } catch { alive = false; } + if (alive) { + lastPid = candidate; + break; + } + } + await new Promise(resolve => setTimeout(resolve, 100)); + } + assert.ok(lastPid, 'fixture subprocess was spawned for the in-flight turn'); + let aliveBefore = true; + try { process.kill(lastPid, 0); } catch { aliveBefore = false; } + assert.ok(aliveBefore, `fixture ${lastPid} is running before shutdown`); + + // Shut the server down while the turn is still running. + child.kill('SIGTERM'); + await new Promise((resolve, reject) => { + const timeout = setTimeout(() => reject(new Error('server did not exit after SIGTERM')), 5_000); + child.on('exit', code => { clearTimeout(timeout); resolve(code); }); + }); + + // The in-flight fixture must be gone — no orphan subprocess may survive. + await new Promise(resolve => setTimeout(resolve, 500)); + let aliveAfter = true; + try { process.kill(lastPid, 0); } catch { aliveAfter = false; } + assert.equal(aliveAfter, false, `fixture process ${lastPid} survived server shutdown`); + await rm(workdir, { recursive: true, force: true }).catch(() => {}); +}, { timeout: 30_000 }); -- 2.43.0