Repository Public
1/**2 * Chaos suite (CONTRACT §2). The run logic (fixtures/chaos-logic.ts, 6 steps: pure, read,3 * idempotent_write, non_idempotent with a probe, non_idempotent without a probe, model_call) runs in a4 * CHILD PROCESS that SIGKILLs itself at one (step, boundary) pair; a fresh worker process then resumes.5 * Matrix: 6 steps × 4 boundaries = 24 kill points, each with its own run and its own effect directory.6 */7import fs from 'node:fs';8import path from 'node:path';9import { afterAll, beforeAll, describe, expect, it } from 'vitest';10import { RECONCILE_OPTIONS, resolveWait } from '../src/index.js';11import { createTestDb, exited, getRun, killAllChildren, newMission, pendingWait, rows, spawnFixture, steps, type RunView, type TestDb } from './helpers.js';1213const BOUNDARIES = ['before_intent', 'after_intent', 'after_exec', 'after_commit'] as const;14type Boundary = (typeof BOUNDARIES)[number];15const UNPROBED = 4; // step s4: non_idempotent without a reconcile probe1617const dbs: TestDb[] = [];18afterAll(async () => {19 await killAllChildren();20 await Promise.all(dbs.map((d) => d.cleanup()));21});22async function freshDb(): Promise<TestDb> {23 const d = await createTestDb();24 dbs.push(d);25 return d;26}2728interface Scenario {29 db: TestDb;30 dir: string;31 missionId: string;32 runId: string;33}34let n = 0;35async function scenario(db: TestDb): Promise<Scenario> {36 const dir = path.join(db.tmp, `chaos-${++n}`);37 fs.mkdirSync(dir, { recursive: true });38 fs.writeFileSync(path.join(dir, 'input.txt'), 'input data');39 const { missionId, rootRunId } = await newMission(db, { workspaceAbs: dir });40 return { db, dir, missionId, runId: rootRunId };41}42/** Runs one worker process until it kills itself or the run is terminal/waiting. */43async function worker(s: Scenario, kill?: { step: number; boundary: Boundary }, extra: Record<string, string> = {}): Promise<{ code: number | null; signal: NodeJS.Signals | null }> {44 const child = spawnFixture('chaos-worker.ts', {45 DATABASE_URL: s.db.url, RUN_ID: s.runId, LEASE_MS: '700',46 ...(kill ? { KILL_STEP: String(kill.step), KILL_BOUNDARY: kill.boundary } : {}),47 ...extra,48 });49 return exited(child);50}51const lines = (file: string): string[] => (fs.existsSync(file) ? fs.readFileSync(file, 'utf8').split('\n').filter(Boolean) : []);52function execCounts(dir: string): number[] {53 const c = [0, 0, 0, 0, 0, 0];54 for (const l of lines(path.join(dir, 'exec.log'))) c[Number(l.slice(1))]!++;55 return c;56}5758/** Invariants that hold after ANY crash once the run is done. */59async function assertInvariants(s: Scenario, run: RunView, reference: unknown): Promise<number[]> {60 expect(run.status).toBe('done');61 expect(run.result?.payload).toEqual(reference); // identical to the uninterrupted run62 const st = await steps(s.db, s.runId);63 expect(st.map((x) => `${x.seq}:${x.name}:${x.status}`)).toEqual(['0:s0:completed', '1:s1:completed', '2:s2:completed', '3:s3:completed', '4:s4:completed', '5:s5:completed']);64 const keys = (await rows<{ idem_key: string }>(s.db, 'SELECT idem_key FROM steps WHERE run_id = $1 ORDER BY seq', [s.runId])).map((r) => r.idem_key);65 const counts = execCounts(s.dir);66 for (const i of [0, 1, 5]) expect(counts[i], `exec count of s${i}`).toBeGreaterThanOrEqual(1); // pure / read / model_call67 expect(fs.readdirSync(path.join(s.dir, 'idem'))).toEqual([keys[2]]); // idempotent_write: observable effect exactly once68 expect(lines(path.join(s.dir, 'ledger.txt'))).toEqual([keys[3]]); // probed non_idempotent: exactly once69 expect(counts[3]).toBe(1);70 expect(lines(path.join(s.dir, 'unprobed.txt'))).toEqual([keys[4]]); // unprobed: exactly once (owner-arbitrated)71 expect(counts[4]).toBe(1);72 // Accounting exactly once.73 expect(run.usage.usd).toBeCloseTo(0.01);74 expect(run.usage.turns).toBe(1);75 expect(Number((await rows<{ spent_usd: string }>(s.db, 'SELECT spent_usd FROM missions WHERE id = $1', [s.missionId]))[0]?.spent_usd)).toBeCloseTo(0.01);76 // No duplicated persisted events.77 const ev = await rows<{ kind: string; seq: number | null; mark: string | null; n: string }>(78 s.db, `SELECT kind, seq, payload->>'mark' AS mark, count(*) AS n FROM events WHERE run_id = $1 GROUP BY 1,2,3`, [s.runId]);79 const cnt = (kind: string, seq: number | null = null, mark: string | null = null): number => Number(ev.find((e) => e.kind === kind && e.seq === seq && e.mark === mark)?.n ?? 0);80 for (let i = 0; i < 6; i++) {81 expect(cnt('step.started', i), `step.started ${i}`).toBe(1);82 expect(cnt('step.completed', i), `step.completed ${i}`).toBe(1);83 }84 for (const mark of ['after-s0', 'after-s3', 'end']) expect(cnt('system.notice', null, mark), `emit ${mark}`).toBe(1);85 expect(cnt('run.usage', 5)).toBe(1);86 expect(ev.filter((e) => e.kind === 'step.failed')).toHaveLength(0);87 return counts;88}8990let reference: unknown;91beforeAll(async () => {92 const db = await freshDb();93 const s = await scenario(db);94 expect(await worker(s)).toEqual({ code: 0, signal: null });95 const run = await getRun(db, s.runId);96 reference = run.result?.payload;97 expect(reference).toEqual({98 a: { v: 42 }, b: { content: 'input data' }, c: { written: true }, d: { ledgered: true }, e: { sent: true }, f: { text: 'ok' },99 });100 expect(await assertInvariants(s, run, reference)).toEqual([1, 1, 1, 1, 1, 1]);101});102103describe.concurrent('chaos: SIGKILL at every step boundary, a fresh worker resumes', () => {104 for (let step = 0; step < 6; step++) {105 it(`step ${step}: ${BOUNDARIES.join(', ')}`, async () => {106 const db = await freshDb();107 for (const boundary of BOUNDARIES) {108 const s = await scenario(db);109 const label = `kill at step ${step} ${boundary}`;110 expect(await worker(s, { step, boundary }), label).toEqual({ code: null, signal: 'SIGKILL' });111 const crashed = await getRun(db, s.runId);112 expect(crashed.status, label).toBe('running'); // died holding the lease113 expect(await worker(s), label).toEqual({ code: 0, signal: null });114 let run = await getRun(db, s.runId);115116 const ambiguous = step === UNPROBED && (boundary === 'after_intent' || boundary === 'after_exec');117 if (ambiguous) {118 // Unprobed non_idempotent found 'started': never retried blindly.119 expect(run.status, label).toBe('waiting');120 expect((await steps(db, s.runId))[UNPROBED]?.status, label).toBe('needs_reconcile');121 const q = await pendingWait(db, s.runId);122 expect(q?.kind).toBe('question');123 expect(q?.step_seq).toBe(UNPROBED);124 expect(q?.payload.options).toEqual([...RECONCILE_OPTIONS]);125 expect(String(q?.payload.text)).toContain('Step 4 (s4) may or may not have run');126 // The owner looks at the world: did the effect happen?127 const happened = lines(path.join(s.dir, 'unprobed.txt')).length === 1;128 expect(happened, label).toBe(boundary === 'after_exec');129 await resolveWait(q!.id, { answer: happened ? RECONCILE_OPTIONS[0] : RECONCILE_OPTIONS[1] }, 'owner', db.k);130 expect(await worker(s), label).toEqual({ code: 0, signal: null });131 run = await getRun(db, s.runId);132 expect(await rows(db, `SELECT 1 FROM events WHERE run_id = $1 AND kind = 'step.needs_reconcile'`, [s.runId])).toHaveLength(1);133 } else {134 expect(await rows(db, `SELECT 1 FROM waits WHERE run_id = $1`, [s.runId]), label).toHaveLength(0);135 }136137 const counts = await assertInvariants(s, run, reference);138 // Killed after exec and before commit: the step is executed again when its class allows it139 // (pure / read / idempotent_write / model_call); the probed and the owner-arbitrated ones are not.140 const twice = boundary === 'after_exec' && [0, 1, 2, 5].includes(step);141 expect(counts[step], label).toBe(twice ? 2 : 1);142 expect(counts.filter((_, i) => i !== step), label).toEqual([1, 1, 1, 1, 1]); // other steps: exactly once143 }144 }, 120_000);145 }146});147148describe('chaos: the three answers to a reconcile question', () => {149 let db: TestDb;150 beforeAll(async () => {151 db = await freshDb();152 });153 async function crashAtUnprobed(boundary: Boundary, extra: Record<string, string> = {}): Promise<{ s: Scenario; waitId: string }> {154 const s = await scenario(db);155 expect(await worker(s, { step: UNPROBED, boundary }, extra)).toEqual({ code: null, signal: 'SIGKILL' });156 expect(await worker(s, undefined, extra)).toEqual({ code: 0, signal: null });157 expect((await getRun(db, s.runId)).status).toBe('waiting');158 const q = await pendingWait(db, s.runId);159 return { s, waitId: q!.id };160 }161162 it('"it ran — continue" without assumedResult records the generic marker and does not re-run the effect', async () => {163 const { s, waitId } = await crashAtUnprobed('after_exec', { CHAOS_ASSUMED: '0' });164 await resolveWait(waitId, { answer: 'it ran — continue' }, 'owner', db.k);165 expect(await worker(s, undefined, { CHAOS_ASSUMED: '0' })).toEqual({ code: 0, signal: null });166 const run = await getRun(db, s.runId);167 expect(run.status).toBe('done');168 expect((run.result?.payload as { e: Record<string, unknown> }).e).toMatchObject({ ok: true, reconciled: 'assumed_done' });169 expect(lines(path.join(s.dir, 'unprobed.txt'))).toHaveLength(1);170 expect(execCounts(s.dir)[UNPROBED]).toBe(1);171 });172173 it('"it did not run — run it" executes the effect once; a second crash during that execution asks again', async () => {174 const { s, waitId } = await crashAtUnprobed('after_intent');175 expect(lines(path.join(s.dir, 'unprobed.txt'))).toHaveLength(0);176 await resolveWait(waitId, { answer: 'it did not run — run it' }, 'owner', db.k);177 // Crash again, this time after the effect and before the commit.178 expect(await worker(s, { step: UNPROBED, boundary: 'after_exec' })).toEqual({ code: null, signal: 'SIGKILL' });179 expect(lines(path.join(s.dir, 'unprobed.txt'))).toHaveLength(1);180 expect(await worker(s)).toEqual({ code: 0, signal: null });181 expect((await getRun(db, s.runId)).status).toBe('waiting'); // asked again, not retried blindly182 const again = await pendingWait(db, s.runId);183 expect(again?.id).not.toBe(waitId);184 await resolveWait(again!.id, { answer: 'it ran — continue' }, 'owner', db.k);185 expect(await worker(s)).toEqual({ code: 0, signal: null });186 await assertInvariants(s, await getRun(db, s.runId), reference);187 });188189 it('"abort" fails the step and ends the run cancelled without running anything else', async () => {190 const { s, waitId } = await crashAtUnprobed('after_intent');191 await resolveWait(waitId, { answer: 'abort' }, 'owner', db.k);192 expect(await worker(s)).toEqual({ code: 0, signal: null });193 const run = await getRun(db, s.runId);194 expect(run.status).toBe('cancelled');195 expect(run.result?.summary).toContain('Aborted by the owner while reconciling step 4');196 const st = await steps(db, s.runId);197 expect(st[UNPROBED]).toMatchObject({ status: 'failed', error: { name: 'ReconcileAborted' } });198 expect(st).toHaveLength(5); // s5 never started199 expect(lines(path.join(s.dir, 'unprobed.txt'))).toHaveLength(0);200 expect((await rows<{ status: string }>(db, 'SELECT status FROM missions WHERE id = $1', [s.missionId]))[0]?.status).toBe('cancelled');201 });202203 it('an answer that is not one of the options asks again', async () => {204 const { s, waitId } = await crashAtUnprobed('after_exec');205 await resolveWait(waitId, { answer: 'dunno' }, 'owner', db.k);206 expect(await worker(s)).toEqual({ code: 0, signal: null });207 expect((await getRun(db, s.runId)).status).toBe('waiting');208 const again = await pendingWait(db, s.runId);209 expect(String(again?.payload.text)).toContain('not one of the options');210 await resolveWait(again!.id, { answer: 'it ran — continue' }, 'owner', db.k);211 expect(await worker(s)).toEqual({ code: 0, signal: null });212 await assertInvariants(s, await getRun(db, s.runId), reference);213 });214});215