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' });