Add bounded inference queue, cancellation, and overload fallback #64

Open
rockachopa wants to merge 3 commits from timmy/17-bounded-inference-queue into main
14 changed files with 972 additions and 26 deletions

View File

@ -4,12 +4,14 @@
"private": true, "private": true,
"type": "module", "type": "module",
"scripts": { "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:ui": "node tests/ui.acceptance.mjs",
"test:photo": "node tests/photo-first.acceptance.mjs", "test:photo": "node tests/photo-first.acceptance.mjs",
"test:sleek": "node tests/sleek-chat.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", "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", "check:diff": "bash scripts/check_diff.sh",
"start": "node server.mjs" "start": "node server.mjs"
}, },

View File

@ -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 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})} 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)=>{ http.createServer(async(req,res)=>{
try{ try{
const url=new URL(req.url,'http://localhost'); 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)} 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'){ 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(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()} if(req.method!=='GET'&&req.method!=='HEAD'){res.writeHead(405,{'allow':'GET, HEAD'});return res.end()}

View File

@ -2,6 +2,7 @@ import { execFile } from 'node:child_process';
import { randomBytes, timingSafeEqual } from 'node:crypto'; import { randomBytes, timingSafeEqual } from 'node:crypto';
import { isAbsolute } from 'node:path'; import { isAbsolute } from 'node:path';
import { detectUrgentText, hasUrgentLedgerContext, urgentChatMessage } from './domain.js'; import { detectUrgentText, hasUrgentLedgerContext, urgentChatMessage } from './domain.js';
import { OverloadError, createInferenceQueue } from './inference-queue.js';
const MAX_MESSAGE_CHARS = 4000; const MAX_MESSAGE_CHARS = 4000;
const MAX_LEDGER_ENTRIES = 20; 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), maxRequestsPerMinute: positiveInt(env.TIMMY_AGENT_RATE_PER_MINUTE, 12, 1, 60),
maxUnlockAttemptsPerMinute: positiveInt(env.TIMMY_AGENT_UNLOCK_RATE_PER_MINUTE, 5, 1, 30), maxUnlockAttemptsPerMinute: positiveInt(env.TIMMY_AGENT_UNLOCK_RATE_PER_MINUTE, 5, 1, 30),
maxSessions: positiveInt(env.TIMMY_AGENT_MAX_SESSIONS, 64, 1, 512), 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) { publicStatus(authenticated = false) {
return { return {
enabled, enabled,
@ -86,12 +89,30 @@ export function parseHermesCliOutput(output) {
return { sessionId: marker[1], reply }; return { sessionId: marker[1], reply };
} }
function execFilePromise(command, args, options) { function execFilePromise(command, args, options, signal) {
return new Promise((resolve, reject) => { 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); if (error) reject(error);
else resolve({ stdout, stderr }); 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; 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]; 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'); if (hermesSessionId) args.push('--resume', hermesSessionId, '--no-restore-cwd');
const output = await execImpl(config.command, args, { const output = await execImpl(config.command, args, {
@ -112,10 +134,15 @@ export async function runHermesCliTurn({ prompt, hermesSessionId, config, execIm
maxBuffer: 1024 * 1024, maxBuffer: 1024 * 1024,
windowsHide: true, windowsHide: true,
env: buildHermesEnvironment(), env: buildHermesEnvironment(),
}); }, signal);
return parseHermesCliOutput(`${output.stderr || ''}\n${output.stdout || ''}`); 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) { function sanitizeLedger(entries) {
if (!Array.isArray(entries)) throw new AgentGatewayError(400, 'Ledger context must be an array.'); if (!Array.isArray(entries)) throw new AgentGatewayError(400, 'Ledger context must be an array.');
return entries.slice(-MAX_LEDGER_ENTRIES).map(entry => ({ return entries.slice(-MAX_LEDGER_ENTRIES).map(entry => ({
@ -139,6 +166,7 @@ export function createHermesAgentService({
runTurn = input => runHermesCliTurn(input), runTurn = input => runHermesCliTurn(input),
randomToken = () => randomBytes(32).toString('base64url'), randomToken = () => randomBytes(32).toString('base64url'),
now = () => Date.now(), now = () => Date.now(),
queue = createInferenceQueue({ concurrency: config.maxConcurrentTurns, maxQueueDepth: config.maxQueueDepth, requestTimeoutMs: config.timeoutMs, now }),
} = {}) { } = {}) {
const sessions = new Map(); const sessions = new Map();
let unlockFailures = []; let unlockFailures = [];
@ -182,7 +210,7 @@ export function createHermesAgentService({
return { cookieToken, public: config.publicStatus(true) }; return { cookieToken, public: config.publicStatus(true) };
}, },
async chat({ origin, cookieToken, payload }) { async chat({ origin, cookieToken, payload, signal }) {
requireOrigin(origin); requireOrigin(origin);
const session = lookup(cookieToken); const session = lookup(cookieToken);
if (!payload || typeof payload !== 'object' || Array.isArray(payload)) throw new AgentGatewayError(400, 'Invalid chat request.'); 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 }; if (urgent) return { reply: urgentChatMessage, connected: true, safetyOverride: true };
const cutoff = now() - RATE_WINDOW_MS; const cutoff = now() - RATE_WINDOW_MS;
session.requests = session.requests.filter(time => time > cutoff); 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.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 { try {
const result = await runTurn({ const result = await (handle = queue.run(async runSignal => {
prompt: buildPrompt(message, ledger, !session.hermesSessionId), if (runSignal.aborted) throw normalizeReason(runSignal.reason);
hermesSessionId: session.hermesSessionId, // Session continuity is read at execution time so a request that
config, // 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 || ''); 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'); 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 }; return { reply, connected: true };
} catch (error) { } catch (error) {
if (error instanceof AgentGatewayError) throw 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.'); 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.');
}

130
src/inference-queue.js Normal file
View File

@ -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 };
}

View File

@ -7,18 +7,22 @@ function providerEndpoint(baseUrl) {
return `${url.toString().replace(/\/$/, '')}/chat/completions`; 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); const photo = validatePhotoPayload(payload);
if (!config?.model) throw new Error('AI analysis is not configured.'); if (!config?.model) throw new Error('AI analysis is not configured.');
const endpoint = providerEndpoint(config.baseUrl); 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, { const response = await fetchImpl(endpoint, {
method: 'POST', method: 'POST',
headers: { headers: {
'content-type': 'application/json', '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 })), 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.'); }); }).catch(() => { throw new Error('AI analysis is temporarily unavailable. Continue manually.'); });
if (!response.ok) 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; let data;

92
tests/agent-queue.test.js Normal file
View 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');
});

View 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/);
});

20
tests/fixtures/slow-hermes.mjs vendored Executable file
View File

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

View File

@ -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; let release;
const runTurn = () => new Promise(resolve => { release = resolve; }); 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' }); await service.unlock({ origin, accessCode: 'test-agent-access-code-2026' });
const active = service.chat({ origin, cookieToken: 'browser-cookie', payload: { message: 'first', ledger: [] } }); const active = service.chat({ origin, cookieToken: 'browser-cookie', payload: { message: 'first', ledger: [] } });
await new Promise(resolve => setImmediate(resolve)); await new Promise(resolve => setImmediate(resolve));
await rejectsStatus(() => service.chat({ origin, cookieToken: 'browser-cookie', payload: { message: 'second', ledger: [] } }), 429); const queued = service.chat({ origin, cookieToken: 'browser-cookie', payload: { message: 'second', ledger: [] } });
release({ reply: 'done', sessionId: 'private-session' }); await new Promise(resolve => setImmediate(resolve));
await active; 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'); } }); 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' }); await broken.unlock({ origin, accessCode: 'test-agent-access-code-2026' });

View File

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

View File

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

View File

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

View File

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

View File

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