Repository Public
1/** Test kit: one throw-away database per test file, in-process workers, polling helpers. */2import { execFileSync, spawn, type ChildProcess } from 'node:child_process';3import { randomBytes } from 'node:crypto';4import fs from 'node:fs';5import os from 'node:os';6import path from 'node:path';7import { fileURLToPath } from 'node:url';8import { createKernel, createMission, migrate, runWorkerLoop, type Kernel, type RunLogic, type WorkerLoopOptions } from '../src/index.js';910export const HERE = path.dirname(fileURLToPath(import.meta.url));11const PKG = path.join(HERE, '..');1213export interface TestDb {14 name: string;15 url: string;16 k: Kernel;17 ownerId: string;18 tmp: string;19 cleanup(): Promise<void>;20}2122export async function createTestDb(): Promise<TestDb> {23 const name = `aos_test_${randomBytes(6).toString('hex')}`;24 execFileSync('createdb', [name], { stdio: 'pipe' });25 const url = `postgres:///${name}?host=/tmp`;26 const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'aos-kernel-test-'));27 try {28 await migrate(url);29 const k = createKernel({ databaseUrl: url, dataDir: path.join(tmp, 'data'), poolMax: 12 });30 const r = await k.pool.query(`INSERT INTO users (username, password_hash) VALUES ('owner', 'x') RETURNING id`);31 const ownerId = (r.rows[0] as { id: string }).id;32 return {33 name, url, k, ownerId, tmp,34 cleanup: async () => {35 await k.close().catch(() => undefined);36 dropDb(name);37 fs.rmSync(tmp, { recursive: true, force: true });38 },39 };40 } catch (err) {41 dropDb(name);42 throw err;43 }44}45function dropDb(name: string): void {46 try {47 execFileSync('dropdb', ['--force', '--if-exists', name], { stdio: 'pipe' });48 } catch {49 /* best effort */50 }51}5253export async function newMission(db: TestDb, over: Record<string, unknown> = {}): Promise<{ missionId: string; rootRunId: string }> {54 return createMission(55 { ownerId: db.ownerId, objective: 'test objective', doneCriteria: ['it works'], provider: 'anthropic', model: 'claude-haiku-5-5', budgetUsd: 5, workspaceAbs: db.tmp, ...over } as Parameters<typeof createMission>[0],56 db.k,57 );58}5960export interface TestWorker {61 stop(): Promise<void>;62}63/** In-process worker with fast timings. */64export function startWorker(db: TestDb, logic: RunLogic, over: Partial<WorkerLoopOptions> = {}): TestWorker {65 const ac = new AbortController();66 const done = runWorkerLoop({67 kernel: db.k, logic, signal: ac.signal, concurrency: 4, leaseMs: 3000, heartbeatMs: 300, pollMs: 50, inlineSleepMs: 100,68 janitorMs: false, dailyBudgetUsd: null, stopGraceMs: 5000, workerId: `w-${randomBytes(3).toString('hex')}`, ...over,69 });70 return {71 stop: async () => {72 ac.abort();73 await done;74 },75 };76}7778export const sleep = (ms: number): Promise<void> => new Promise((r) => setTimeout(r, ms));7980export async function waitFor<T>(fn: () => Promise<T | null | undefined | false>, timeoutMs = 20_000, what = 'condition'): Promise<T> {81 const end = Date.now() + timeoutMs;82 for (;;) {83 const v = await fn();84 if (v) return v;85 if (Date.now() > end) throw new Error(`timeout waiting for ${what}`);86 await sleep(25);87 }88}8990export interface RunView {91 status: string;92 result: { status: string; summary: string; payload: unknown } | null;93 usage: Record<string, number>;94 fencing_token: string;95 cancel_requested: boolean;96 lease_owner: string | null;97 next_seq: number;98}99export async function getRun(db: TestDb, runId: string): Promise<RunView> {100 const r = await db.k.pool.query('SELECT status, result, usage, fencing_token::text, cancel_requested, lease_owner, next_seq FROM runs WHERE id = $1', [runId]);101 return r.rows[0] as RunView;102}103export async function waitStatus(db: TestDb, runId: string, statuses: string[], timeoutMs = 20_000): Promise<RunView> {104 return waitFor(async () => {105 const r = await getRun(db, runId);106 return statuses.includes(r.status) ? r : null;107 }, timeoutMs, `run ${runId} to be ${statuses.join('|')}`);108}109export async function rows<T = Record<string, unknown>>(db: TestDb, sql: string, params: unknown[] = []): Promise<T[]> {110 return (await db.k.pool.query(sql, params)).rows as T[];111}112export async function steps(db: TestDb, runId: string): Promise<{ seq: number; kind: string; name: string; status: string; result: unknown; error: { name: string } | null }[]> {113 return rows(db, 'SELECT seq, kind, name, status, result, error FROM steps WHERE run_id = $1 ORDER BY seq', [runId]);114}115export async function pendingWait(db: TestDb, runId: string): Promise<{ id: string; kind: string; payload: Record<string, unknown>; step_seq: number } | null> {116 const r = await rows<{ id: string; kind: string; payload: Record<string, unknown>; step_seq: number }>(117 db, `SELECT id, kind, payload, step_seq FROM waits WHERE run_id = $1 AND status = 'pending' ORDER BY created_at LIMIT 1`, [runId]);118 return r[0] ?? null;119}120121// ---- child processes -------------------------------------------------------------------------122const children = new Set<ChildProcess>();123export function spawnFixture(file: string, env: Record<string, string>): ChildProcess {124 // `node --import tsx` keeps the fixture in ONE process (the tsx CLI would fork a child that survives a SIGKILL).125 const child = spawn(process.execPath, ['--import', 'tsx', path.join(HERE, 'fixtures', file)], {126 cwd: PKG,127 env: { ...process.env, AOS_LOG_LEVEL: 'silent', ...env },128 stdio: ['ignore', 'ignore', 'inherit'],129 });130 children.add(child);131 child.once('exit', () => children.delete(child));132 return child;133}134export function exited(child: ChildProcess): Promise<{ code: number | null; signal: NodeJS.Signals | null }> {135 if (child.exitCode !== null || child.signalCode !== null) return Promise.resolve({ code: child.exitCode, signal: child.signalCode });136 return new Promise((resolve) => child.once('exit', (code, signal) => resolve({ code, signal })));137}138export async function killAllChildren(): Promise<void> {139 const all = [...children];140 for (const c of all) c.kill('SIGKILL');141 await Promise.all(all.map(exited));142}143