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.
This commit is contained in:
parent
38a59c8ee9
commit
5dcaaae4d6
|
|
@ -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.');
|
||||
}
|
||||
|
|
|
|||
92
tests/agent-queue.test.js
Normal file
92
tests/agent-queue.test.js
Normal file
|
|
@ -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');
|
||||
});
|
||||
109
tests/agent-turn-timeout.test.js
Normal file
109
tests/agent-turn-timeout.test.js
Normal file
|
|
@ -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/);
|
||||
});
|
||||
|
|
@ -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' });
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user