Repository Public
1/**2 * Crash / replay (CONTRACT §5 "Replay rule"). The worker runs in a CHILD PROCESS that SIGKILLs itself at3 * one (step, boundary) of a 7-turn scripted mission — in the middle of a parallel tool batch and at a4 * dozen other points. A fresh process resumes. Each scenario is compared with an uninterrupted run:5 * final transcript shown to the model, files on disk, git history, journal shape and result envelope.6 */7import fs from 'node:fs';8import path from 'node:path';9import { RECONCILE_OPTIONS, resolveWait } from '@aos/kernel';10import { afterAll, beforeAll, describe, expect, it } from 'vitest';11import { readRequestLog, toolResultsOf, type FakeEntry } from '../src/testkit/index.js';12import { SCENARIO_ANSWER, SCENARIO_CRITERIA, SCENARIO_FILES } from './fixtures/scenario.js';13import {14 createTestDb, criteriaOf, exited, getRun, git, killAllChildren, makeWorkspace, newMission, rows, spawnFixture, steps, TERMINAL,15 type MissionRef, type StepView, type TestDb,16} from './helpers.js';1718type Boundary = 'before_intent' | 'after_intent' | 'after_exec' | 'after_commit';19interface Kill {20 seq: number;21 boundary: Boundary;22}23interface Outcome {24 m: MissionRef;25 kills: number;26 spawns: number;27 log: string;28}2930let db: TestDb;31beforeAll(async () => {32 db = await createTestDb();33});34afterAll(async () => {35 await killAllChildren();36 await db.cleanup();37});3839let n = 0;40/** Runs the scenario to its end with as many worker processes as it takes; `kill` is armed until one dies. */41async function drive(kill?: Kill): Promise<Outcome> {42 const ws = makeWorkspace(db, { git: true, files: SCENARIO_FILES });43 const m = await newMission(db, ws, { doneCriteria: SCENARIO_CRITERIA, objective: 'Fix add() in src/calc.js' });44 const log = path.join(db.tmp, `requests-${++n}.jsonl`);45 let armed = kill;46 let kills = 0;47 let spawns = 0;48 for (; spawns < 12; ) {49 spawns++;50 const child = spawnFixture('crash-worker.ts', {51 DATABASE_URL: db.url, DATA_DIR: db.k.dataDir, WORKSPACE_ROOT: db.root, RUN_ID: m.runId, LOG_FILE: log, LEASE_MS: '700',52 ...(armed ? { KILL_SEQ: String(armed.seq), KILL_BOUNDARY: armed.boundary } : {}),53 });54 const { signal, code } = await exited(child);55 if (signal === 'SIGKILL') {56 kills++;57 armed = undefined;58 continue;59 }60 expect(code, 'worker exit code').toBe(0);61 const run = await getRun(db, m.runId);62 if (TERMINAL.includes(run.status)) return { m, kills, spawns, log };63 expect(run.status).toBe('waiting');64 const w = (await rows<{ id: string; payload: { reconcile?: boolean } }>(db, `SELECT id, payload FROM waits WHERE run_id = $1 AND status = 'pending'`, [m.runId]))[0];65 expect(w?.payload.reconcile, 'no step may need owner reconciliation in this scenario').toBeUndefined();66 await resolveWait(w?.id as string, { answer: SCENARIO_ANSWER }, 'owner', db.k);67 }68 throw new Error('scenario did not finish in 12 worker processes');69}7071/** Commit hashes, idempotency keys and the mission branch differ between two missions: mask them. */72const mask = (s: string): string => s.replace(/\b[0-9a-f]{7,64}\b/g, '<hex>').replace(/aos\/<hex>|aos\/[0-9a-f]{8}/g, 'aos/<mission>');73function norm<T>(v: T): T {74 const walk = (x: unknown): unknown => {75 if (typeof x === 'string') return mask(x);76 if (Array.isArray(x)) return x.map(walk);77 if (x && typeof x === 'object') return Object.fromEntries(Object.entries(x).map(([k, val]) => [k, walk(val)]));78 return x;79 };80 return walk(v) as T;81}8283/** Crash-recovery wording of the write tools: same effect, different sentence than the first execution. */84const RECOVERED: Record<string, RegExp> = {85 write_file: /already has this content \(no change\)/,86 edit_file: /\(already applied by this step\)/,87 git_commit: /^Already committed by this step/,88};8990function tree(ws: string): Record<string, string> {91 const out: Record<string, string> = {};92 const walk = (dir: string): void => {93 for (const e of fs.readdirSync(dir, { withFileTypes: true })) {94 if (e.name === '.git') continue;95 const p = path.join(dir, e.name);96 if (e.isDirectory()) walk(p);97 else out[path.relative(ws, p)] = fs.readFileSync(p, 'utf8');98 }99 };100 walk(ws);101 return out;102}103const shape = (st: StepView[]): string[] => st.map((s) => `${s.seq}:${s.kind}/${s.name}:${s.status}`);104const finalTranscript = (log: string): FakeEntry[] => readRequestLog(log).at(-1)?.transcript ?? [];105106describe('crash and replay', () => {107 let ref: Outcome;108 let refSteps: StepView[];109 let refTranscript: FakeEntry[];110111 it('reference: the uninterrupted run', async () => {112 ref = await drive();113 refSteps = await steps(db, ref.m.runId);114 refTranscript = finalTranscript(ref.log);115 const run = await getRun(db, ref.m.runId);116 expect(run.status).toBe('done');117 expect(ref.kills).toBe(0);118 expect(ref.spawns).toBe(2); // one up to the owner's question, one after the answer119 expect(tree(ref.m.ws)).toEqual({120 'a.txt': 'alpha\n', 'counter.log': 'tick\n', 'notes.md': '# Notes\nadd() subtracted instead of adding.\n', 'src/calc.js': 'export const add = (a, b) => a + b;\n',121 });122 expect(git(ref.m.ws, 'log', '--format=%s').trim().split('\n')).toEqual(['Fix add()', 'init']);123 expect(git(ref.m.ws, 'status', '--porcelain')).toBe('');124 expect(refSteps.every((s) => s.status === 'completed')).toBe(true);125 expect(toolResultsOf(refTranscript).map((r) => `${r.name}:${r.ok}`)).toEqual([126 'read_file:true', 'read_file:true', 'list_dir:true', 'write_file:true', 'edit_file:true', 'shell:true', 'git_commit:true', 'ask_owner:true', 'update_criteria:true', 'git_status:true',127 ]);128 expect(toolResultsOf(refTranscript)[5]?.content).toBe('[exit code 0]\n1');129 });130131 /** nth step (0-based) of a kind/name in the reference journal. */132 const seqOf = (kind: string, name: string | null, nth = 0): number => {133 const s = refSteps.filter((x) => x.kind === kind && (name === null || x.name === name))[nth];134 if (!s) throw new Error(`no step ${kind}/${name ?? '*'} #${nth} in the reference`);135 return s.seq;136 };137 const POINTS: [string, () => Kill][] = [138 ['middle of the parallel read batch (2nd of 3 reads executed, not committed)', () => ({ seq: seqOf('tool_call', 'read_file', 1), boundary: 'after_exec' })],139 ['parallel read batch, first read committed', () => ({ seq: seqOf('tool_call', 'read_file', 0), boundary: 'after_commit' })],140 ['parallel read batch, last call before its intent', () => ({ seq: seqOf('tool_call', 'list_dir'), boundary: 'before_intent' })],141 ['setup: prefix executed, not committed', () => ({ seq: seqOf('setup', 'prefix'), boundary: 'after_exec' })],142 ['setup: mission branch created, not committed', () => ({ seq: seqOf('setup', 'git_branch'), boundary: 'after_exec' })],143 ['first model call after its intent', () => ({ seq: seqOf('model_call', null, 0), boundary: 'after_intent' })],144 ['third model call answered, not committed', () => ({ seq: seqOf('model_call', null, 2), boundary: 'after_exec' })],145 ['policy snapshot of the write batch executed, not committed', () => ({ seq: seqOf('policy', 'policy_rules', 1), boundary: 'after_exec' })],146 ['write_file written, not committed', () => ({ seq: seqOf('tool_call', 'write_file'), boundary: 'after_exec' })],147 ['edit_file between the two writes of the batch (intent only)', () => ({ seq: seqOf('tool_call', 'edit_file'), boundary: 'after_intent' })],148 ['edit_file applied, not committed', () => ({ seq: seqOf('tool_call', 'edit_file'), boundary: 'after_exec' })],149 ['shell after its intent (not executed yet)', () => ({ seq: seqOf('tool_call', 'shell'), boundary: 'after_intent' })],150 ['shell committed', () => ({ seq: seqOf('tool_call', 'shell'), boundary: 'after_commit' })],151 ['git_commit done, not committed to the journal', () => ({ seq: seqOf('tool_call', 'git_commit'), boundary: 'after_exec' })],152 ['clock reading before the owner question', () => ({ seq: seqOf('now', null, 0), boundary: 'after_commit' })],153 ['after the answer: update_criteria stored, not committed', () => ({ seq: seqOf('tool_call', 'update_criteria'), boundary: 'after_exec' })],154 ['finish gate recorded', () => ({ seq: seqOf('tool_call', 'finish'), boundary: 'after_commit' })],155 ];156157 it.each(POINTS)('kill at %s → identical outcome', async (_label, point) => {158 const kill = point();159 const out = await drive(kill);160 expect(out.kills, 'the worker must really have been killed').toBe(1);161 const run = await getRun(db, out.m.runId);162 const refRun = await getRun(db, ref.m.runId);163 expect(run.status).toBe('done');164 expect(norm(run.result)).toEqual(norm(refRun.result));165 expect(run.usage.turns).toBeGreaterThanOrEqual(refRun.usage.turns ?? 0);166167 // journal: same steps, same order, all completed (seqs are deterministic across the crash)168 const st = await steps(db, out.m.runId);169 expect(shape(st)).toEqual(shape(refSteps));170171 // files on disk, git history, criteria172 expect(tree(out.m.ws)).toEqual(tree(ref.m.ws));173 expect(fs.readFileSync(path.join(out.m.ws, 'counter.log'), 'utf8')).toBe('tick\n'); // the shell command ran exactly once174 expect(git(out.m.ws, 'log', '--format=%s').trim().split('\n')).toEqual(['Fix add()', 'init']); // exactly one commit175 expect(git(out.m.ws, 'status', '--porcelain')).toBe('');176 expect(git(out.m.ws, 'show', '--stat', '--format=', 'HEAD').trim()).toBe(git(ref.m.ws, 'show', '--stat', '--format=', 'HEAD').trim());177 expect(await criteriaOf(db, out.m.missionId)).toEqual(await criteriaOf(db, ref.m.missionId));178179 // the final transcript shown to the model180 const got = norm(finalTranscript(out.log));181 const want = norm(refTranscript);182 const victim = refSteps.find((s) => s.seq === kill.seq) as StepView;183 const recovered = kill.boundary === 'after_exec' && victim.kind === 'tool_call' ? RECOVERED[victim.name] : undefined;184 if (!recovered) {185 expect(got).toEqual(want);186 } else {187 // The effect had landed before the crash; the re-executed tool recognises it and says so. Everything188 // else in the transcript is identical.189 const a = toolResultsOf(got);190 const b = toolResultsOf(want);191 const idx = b.findIndex((r) => r.name === victim.name);192 expect(a[idx]?.ok).toBe(true);193 expect(a[idx]?.content).toMatch(recovered);194 const patch = (t: FakeEntry[]): FakeEntry[] => t.map((e) => (e.role === 'tool' ? { ...e, results: e.results.map((r) => (r.name === victim.name ? { ...r, content: '<result>' } : r)) } : e));195 expect(patch(got)).toEqual(patch(want));196 }197 // every transcript any worker process sent is a prefix-consistent view of the final one198 for (const req of readRequestLog(out.log)) {199 const t = norm(req.transcript);200 expect(got.slice(0, t.length).map((e) => e.role)).toEqual(t.map((e) => e.role));201 }202 }, 90_000);203204 /** One worker process on an existing mission; returns how it ended. */205 async function worker(m: MissionRef, scenario: string, log: string, kill?: Kill): Promise<NodeJS.Signals | null> {206 const child = spawnFixture('crash-worker.ts', {207 DATABASE_URL: db.url, DATA_DIR: db.k.dataDir, WORKSPACE_ROOT: db.root, RUN_ID: m.runId, LOG_FILE: log, LEASE_MS: '700', SCENARIO: scenario,208 ...(kill ? { KILL_SEQ: String(kill.seq), KILL_BOUNDARY: kill.boundary } : {}),209 });210 return (await exited(child)).signal;211 }212 const pending = async (m: MissionRef) => (await rows<{ id: string; kind: string; payload: Record<string, unknown> }>(db, `SELECT id, kind, payload FROM waits WHERE run_id = $1 AND status = 'pending'`, [m.runId]))[0];213214 it('a push killed after it reached the remote is reconciled by its probe: pushed exactly once, no question', async () => {215 const ws = makeWorkspace(db, { git: true, files: { 'a.txt': 'a\n' } });216 const remote = path.join(db.tmp, `remote-${path.basename(ws)}.git`);217 git(db.tmp, 'init', '-q', '--bare', remote);218 git(ws, 'remote', 'add', 'origin', remote);219 const m = await newMission(db, ws);220 const log = path.join(db.tmp, `requests-push-${++n}.jsonl`);221 expect(await worker(m, 'push', log)).toBeNull(); // runs up to the approval222 const approval = await pending(m);223 expect(approval?.kind).toBe('approval');224 await resolveWait(approval?.id as string, { decision: 'approve' }, 'owner', db.k);225 const pushSeq = ((await steps(db, m.runId)).at(-1)?.seq ?? 0) + 2; // the journal ends with the approval wait; then a clock reading, then the push226 expect(await worker(m, 'push', log, { seq: pushSeq, boundary: 'after_exec' })).toBe('SIGKILL');227 const branch = `aos/${m.missionId.slice(0, 8)}`;228 expect(git(remote, 'branch', '--format=%(refname:short)').trim()).toBe(branch); // the effect happened before the crash229 const pushedSha = git(remote, 'rev-parse', branch).trim();230 expect(await worker(m, 'push', log)).toBeNull();231 const run = await getRun(db, m.runId);232 expect(run.status).toBe('partial');233 expect(await pending(m)).toBeUndefined();234 const st = await steps(db, m.runId);235 const push = st.find((s) => s.name === 'git_push');236 expect(push).toMatchObject({ seq: pushSeq, status: 'completed', effect: 'non_idempotent', result: { ok: true } });237 expect(git(remote, 'rev-parse', branch).trim()).toBe(pushedSha);238 expect(toolResultsOf(finalTranscript(log))[0]).toMatchObject({ name: 'git_push', ok: true });239 }, 90_000);240241 it('a network-enabled shell command killed mid-flight is never retried blindly: the owner arbitrates', async () => {242 const ws = makeWorkspace(db, { files: { 'a.txt': 'a\n' } });243 await db.k.pool.query(`INSERT INTO policy_rules (owner_id, project_id, decision, tool, match) VALUES ($1, $2, 'allow', 'shell', '{}')`, [db.ownerId, db.projectId]);244 const log = path.join(db.tmp, `requests-net-${++n}.jsonl`);245 // an uninterrupted run first, to learn the seq of the shell step (one mission at a time: a worker claims the oldest queued run)246 const ref = await newMission(db, makeWorkspace(db, { files: { 'a.txt': 'a\n' } }), { network: 'open' });247 expect(await worker(ref, 'netshell', path.join(db.tmp, `requests-net-ref-${n}.jsonl`))).toBeNull();248 const m = await newMission(db, ws, { network: 'open' });249 const shellSeq = (await steps(db, ref.runId)).find((s) => s.name === 'shell')?.seq as number;250 expect((await steps(db, ref.runId)).find((s) => s.name === 'shell')?.effect).toBe('non_idempotent');251252 expect(await worker(m, 'netshell', log, { seq: shellSeq, boundary: 'after_exec' })).toBe('SIGKILL');253 expect(fs.readFileSync(path.join(ws, 'counter.log'), 'utf8')).toBe('tick\n');254 expect(await worker(m, 'netshell', log)).toBeNull(); // suspends on the reconcile question255 const q = await pending(m);256 expect(q).toMatchObject({ kind: 'question', payload: { reconcile: true, stepName: 'shell', options: [...RECONCILE_OPTIONS] } });257 expect((await steps(db, m.runId)).find((s) => s.name === 'shell')?.status).toBe('needs_reconcile');258 await resolveWait(q?.id as string, { answer: RECONCILE_OPTIONS[0] }, 'owner', db.k);259 expect(await worker(m, 'netshell', log)).toBeNull();260 expect((await getRun(db, m.runId)).status).toBe('partial');261 expect(fs.readFileSync(path.join(ws, 'counter.log'), 'utf8')).toBe('tick\n'); // not run a second time262 const seen = toolResultsOf(finalTranscript(log))[0];263 expect(seen).toMatchObject({ name: 'shell', ok: true });264 expect(seen?.content).toMatch(/owner confirmed that this call had already run before a crash/);265 await db.k.pool.query('DELETE FROM policy_rules');266 }, 90_000);267});268