timmy-talking-turd/tests/inference-queue.acceptance.mjs

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