Repository Public
1import { createHash } from 'node:crypto';2import { afterAll, beforeAll, describe, expect, it } from 'vitest';3import type { AosEvent } from '@aos/shared';4import {5 appendEvent, canonicalJson, KernelStorageError, listenEvents, migrate, openArtifact, putArtifact, StepError,6 type RunContext, type RunLogic,7} from '../src/index.js';8import { createTestDb, getRun, newMission, rows, sleep, startWorker, steps, waitFor, waitStatus, type TestDb } from './helpers.js';910let db: TestDb;11beforeAll(async () => {12 db = await createTestDb();13});14afterAll(async () => {15 await db.cleanup();16});1718const done = (payload: unknown = null) => ({ status: 'done' as const, summary: 'ok', payload, artifacts: [], evidence: [], openQuestions: [] });19const kinds = async (missionId: string): Promise<string[]> =>20 (await rows<{ kind: string }>(db, 'SELECT kind FROM events WHERE mission_id = $1 ORDER BY id', [missionId])).map((e) => e.kind);2122describe('happy path', () => {23 it('runs a mission to completion and journals every step', async () => {24 const { missionId, rootRunId } = await newMission(db);25 expect((await getRun(db, rootRunId)).status).toBe('queued');26 const logic: RunLogic = async (ctx) => {27 const a = await ctx.step({ kind: 'calc', name: 'add', effect: 'pure', request: { a: 1, b: 2 } }, async () => ({ result: { sum: 3 } }));28 const b = await ctx.step({ kind: 'model_call', name: 'm', effect: 'model_call', request: { prompt: 'hi' } }, async (idem) => ({29 result: { text: 'hello', idemLen: idem.length },30 usage: { usd: 0.02, inputTokens: 100, outputTokens: 20 },31 }));32 const t = await ctx.now();33 const u = await ctx.uuid();34 return done({ a, b, t: t.toISOString(), u, usage: ctx.usage(), left: ctx.budgetLeft().usd });35 };36 const w = startWorker(db, logic);37 const run = await waitStatus(db, rootRunId, ['done']);38 await w.stop();39 const p = run.result?.payload as { a: { sum: number }; b: { text: string; idemLen: number }; u: string; usage: { usd: number; turns: number }; left: number };40 expect(p.a).toEqual({ sum: 3 });41 expect(p.b).toEqual({ text: 'hello', idemLen: 64 });42 expect(p.u).toMatch(/^[0-9a-f-]{36}$/);43 expect(p.usage.usd).toBeCloseTo(0.02);44 expect(p.usage.turns).toBe(1);45 expect(p.left).toBeCloseTo(4.98);46 expect(run.usage.usd).toBeCloseTo(0.02);47 expect(run.next_seq).toBe(4);48 const st = await steps(db, rootRunId);49 expect(st.map((s) => [s.seq, s.kind, s.status])).toEqual([[0, 'calc', 'completed'], [1, 'model_call', 'completed'], [2, 'now', 'completed'], [3, 'uuid', 'completed']]);50 const m = (await rows<{ status: string; spent_usd: string; ended_at: Date | null; started_at: Date | null }>(db, 'SELECT status, spent_usd, started_at, ended_at FROM missions WHERE id = $1', [missionId]))[0];51 expect(m?.status).toBe('done');52 expect(Number(m?.spent_usd)).toBeCloseTo(0.02);53 expect(m?.started_at).not.toBeNull();54 expect(m?.ended_at).not.toBeNull();55 expect(await kinds(missionId)).toEqual([56 'mission.created', 'run.created', 'run.status', 'mission.status',57 'step.started', 'step.completed', 'step.started', 'step.completed', 'run.usage',58 'run.status', 'mission.status',59 ]);60 });6162 it('an uncaught logic error fails the run with its message; an invalid envelope fails it too', async () => {63 const a = await newMission(db);64 const b = await newMission(db);65 const w = startWorker(db, async (ctx) => {66 if (ctx.runId === a.rootRunId) throw new Error('boom');67 return { nope: true } as never;68 });69 const ra = await waitStatus(db, a.rootRunId, ['failed']);70 const rb = await waitStatus(db, b.rootRunId, ['failed']);71 await w.stop();72 expect(ra.result?.summary).toContain('boom');73 expect(rb.result?.summary).toContain('invalid result envelope');74 });75});7677describe('replay', () => {78 it('returns recorded results without re-executing (exec counters) and keeps now()/uuid() stable', async () => {79 const { rootRunId } = await newMission(db);80 const count = { a: 0, b: 0, logic: 0 };81 const seen: string[] = [];82 const replayFlags: boolean[] = [];83 const logic: RunLogic = async (ctx) => {84 count.logic++;85 replayFlags.push(ctx.replaying);86 const a = await ctx.step({ kind: 'calc', name: 'a', effect: 'pure', request: {} }, async () => ({ result: { n: ++count.a, nul: 'x\u0000y', z: 1, a: 2 } }));87 const t = await ctx.now();88 const u = await ctx.uuid();89 seen.push(`${canonicalJson(a)}|${Object.keys(a).join(',')}|${t.toISOString()}|${u}`);90 await ctx.sleep(300); // > inlineSleepMs ⇒ the run is suspended and replayed from the top91 const b = await ctx.step({ kind: 'calc', name: 'b', effect: 'pure', request: {} }, async () => ({ result: { n: ++count.b } }));92 return done({ a, b });93 };94 const w = startWorker(db, logic);95 const run = await waitStatus(db, rootRunId, ['done']);96 await w.stop();97 expect(count).toEqual({ a: 1, b: 1, logic: 2 });98 expect(seen).toHaveLength(2);99 expect(seen[0]).toBe(seen[1]); // same bytes AND same key order on first execution and on replay100 expect(seen[0]).toContain('x�y'); // NUL is not storable in jsonb: replaced consistently101 expect(replayFlags).toEqual([false, true]);102 expect((run.result?.payload as { a: { n: number } }).a.n).toBe(1);103 });104105 it('request-hash mismatch ⇒ NonDeterminismError ⇒ run failed', async () => {106 const { rootRunId } = await newMission(db);107 let invocation = 0;108 const w = startWorker(db, async (ctx) => {109 const n = ++invocation; // non-deterministic on purpose110 await ctx.step({ kind: 'calc', name: 'a', effect: 'pure', request: { n } }, async () => ({ result: {} }));111 await ctx.sleep(300);112 return done();113 });114 const run = await waitStatus(db, rootRunId, ['failed', 'done']);115 await w.stop();116 expect(run.status).toBe('failed');117 expect(run.result?.summary).toMatch(/Non-deterministic run logic at step 0.*request hash differs/);118 });119120 it('a different step kind at the same seq, or returning before the journal ends, is non-determinism too', async () => {121 const a = await newMission(db);122 const b = await newMission(db);123 const inv = new Map<string, number>();124 const w = startWorker(db, async (ctx) => {125 const n = (inv.get(ctx.runId) ?? 0) + 1;126 inv.set(ctx.runId, n);127 if (ctx.runId === a.rootRunId) {128 if (n === 1) await ctx.step({ kind: 'calc', name: 'a', effect: 'pure', request: {} }, async () => ({ result: {} }));129 else await ctx.now();130 await ctx.sleep(300);131 } else if (n === 1) {132 await ctx.now();133 await ctx.sleep(300);134 }135 return done();136 });137 const ra = await waitStatus(db, a.rootRunId, ['failed', 'done']);138 const rb = await waitStatus(db, b.rootRunId, ['failed', 'done']);139 await w.stop();140 expect(ra.result?.summary).toMatch(/journal has calc\/a/);141 expect(rb.result?.summary).toMatch(/journal has more/);142 });143144 it('a failed step is journaled and rethrown identically on replay, without re-executing', async () => {145 const { rootRunId } = await newMission(db);146 let execs = 0;147 const caught: string[] = [];148 const w = startWorker(db, async (ctx) => {149 try {150 await ctx.step({ kind: 'tool_call', name: 'flaky', effect: 'read', request: {} }, async () => {151 execs++;152 throw Object.assign(new TypeError('no such thing'), { code: 'ENOENT', status: 404, retryable: false });153 });154 } catch (err) {155 const e = err as StepError;156 caught.push(`${e instanceof StepError}|${e.name}|${e.errorName}|${e.message}|${e.code}|${e.status}|${e.retryable}|${e.stepSeq}`);157 }158 await ctx.sleep(300);159 return done();160 });161 await waitStatus(db, rootRunId, ['done']);162 await w.stop();163 expect(execs).toBe(1);164 expect(caught).toEqual(Array(2).fill('true|StepError|TypeError|no such thing|ENOENT|404|false|0'));165 expect((await steps(db, rootRunId))[0]?.status).toBe('failed');166 });167168 it('parallel steps get their seq synchronously in call order, stable across replays', async () => {169 const { rootRunId } = await newMission(db);170 const execs: number[] = [];171 const w = startWorker(db, async (ctx) => {172 const rs = await Promise.all(173 [0, 1, 2, 3, 4, 5].map((i) =>174 ctx.step({ kind: 'tool_call', name: `r${i}`, effect: 'read', request: { i } }, async () => {175 execs.push(i);176 await sleep(60 - i * 10); // later calls finish first177 return { result: { i } };178 }),179 ),180 );181 await ctx.sleep(300);182 const after = await ctx.step({ kind: 'calc', name: 'after', effect: 'pure', request: {} }, async () => ({ result: {} }));183 return done({ rs, after });184 });185 const run = await waitStatus(db, rootRunId, ['done', 'failed']);186 await w.stop();187 expect(run.status).toBe('done');188 expect(execs.sort()).toEqual([0, 1, 2, 3, 4, 5]); // each executed once189 const st = await steps(db, rootRunId);190 expect(st.map((s) => s.name)).toEqual(['r0', 'r1', 'r2', 'r3', 'r4', 'r5', 'sleep', 'after']);191 expect((run.result?.payload as { rs: { i: number }[] }).rs.map((r) => r.i)).toEqual([0, 1, 2, 3, 4, 5]);192 });193194 it('emit() is exactly-once across replays (keyed by emit index)', async () => {195 const { missionId, rootRunId } = await newMission(db);196 const w = startWorker(db, async (ctx) => {197 await ctx.emit('policy.decision', { n: 1 });198 await ctx.emit('system.notice', { n: 2 }, 'system');199 await ctx.sleep(300);200 await ctx.emit('policy.decision', { n: 3 });201 await ctx.sleep(300);202 await ctx.emit('policy.decision', { n: 4 });203 return done();204 });205 await waitStatus(db, rootRunId, ['done']);206 await w.stop();207 const ev = await rows<{ kind: string; trust: string; payload: { n: number } }>(208 db, `SELECT kind, trust, payload FROM events WHERE mission_id = $1 AND kind IN ('policy.decision','system.notice') ORDER BY id`, [missionId]);209 expect(ev.map((e) => e.payload.n)).toEqual([1, 2, 3, 4]);210 expect(ev.map((e) => e.trust)).toEqual(['agent', 'system', 'agent', 'agent']);211 });212213 it('exec finished but the commit failed ⇒ the step is found started and re-executed on the next claim', async () => {214 const { rootRunId } = await newMission(db);215 let execs = 0;216 let failed = false;217 const w = startWorker(218 db,219 async (ctx) => {220 const r = await ctx.step({ kind: 'tool_call', name: 'w', effect: 'idempotent_write', request: {} }, async () => ({ result: { n: ++execs } }));221 return done(r);222 },223 {224 hooks: {225 boundary: (i) => {226 if (i.runId === rootRunId && i.boundary === 'after_exec' && !failed) {227 failed = true;228 throw new KernelStorageError('simulated commit failure');229 }230 },231 },232 },233 );234 const run = await waitStatus(db, rootRunId, ['done', 'failed']);235 await w.stop();236 expect(run.status).toBe('done');237 expect(execs).toBe(2);238 expect(run.result?.payload).toEqual({ n: 2 });239 expect(Number(run.fencing_token)).toBe(2); // claimed twice240 });241});242243describe('logic with a broad catch', () => {244 it('cannot swallow a suspension, a cancel or a non-determinism error', async () => {245 const a = await newMission(db); // swallows Suspend246 const b = await newMission(db); // swallows NonDeterminismError247 const inv = new Map<string, number>();248 const after = { a: 0 };249 const w = startWorker(db, async (ctx) => {250 const n = (inv.get(ctx.runId) ?? 0) + 1;251 inv.set(ctx.runId, n);252 if (ctx.runId === a.rootRunId) {253 let slept = true;254 try {255 await ctx.sleep(300);256 } catch {257 slept = false; // swallowed258 }259 // On the first invocation the run is suspended: this step must NOT execute.260 await ctx.step({ kind: 'calc', name: 'after', effect: 'pure', request: {} }, async () => ({ result: { n: ++after.a } })).catch(() => undefined);261 return done({ slept });262 }263 try {264 await ctx.step({ kind: 'calc', name: 'x', effect: 'pure', request: { n } }, async () => ({ result: {} }));265 } catch {266 /* swallowed */267 }268 if (n === 1) await ctx.sleep(300);269 return done();270 });271 const ra = await waitStatus(db, a.rootRunId, ['done', 'failed']);272 const rb = await waitStatus(db, b.rootRunId, ['done', 'failed']);273 await w.stop();274 expect(ra.status).toBe('done');275 expect(ra.result?.payload).toEqual({ slept: true }); // the result of the replay, not of the swallowed attempt276 expect(after.a).toBe(1);277 expect(rb.status).toBe('failed');278 expect(rb.result?.summary).toMatch(/Non-deterministic/);279 });280});281282describe('events', () => {283 it('listenEvents delivers persisted events once, in order, from a cursor; transient deltas with id -1', async () => {284 const { missionId, rootRunId } = await newMission(db);285 const before = await appendEvent({ missionId, kind: 'system.notice', payload: { n: 'before' } }, db.k);286 const got: AosEvent[] = [];287 const sub = listenEvents((ev) => void (ev.missionId === missionId && got.push(ev)), { kernel: db.k, cursor: before - 1, pollMs: 200 });288 await sub.ready;289 const big = 'é€😀x'.repeat(6000); // ≈ 60 kB of UTF-8: must be chunked under the NOTIFY limit290 const w = startWorker(db, async (ctx: RunContext) => {291 await ctx.step({ kind: 'model_call', name: 'm', effect: 'model_call', request: {} }, async () => {292 ctx.emitTransient('model.delta', { type: 'text', text: 'small' });293 ctx.emitTransient('model.delta', { type: 'text', text: big });294 ctx.emitTransient('model.delta', { type: 'blob', data: 'y'.repeat(20_000) }); // cannot be chunked: dropped295 return { result: {}, usage: { usd: 0.001 } };296 });297 return done();298 });299 await waitStatus(db, rootRunId, ['done']);300 await waitFor(async () => got.some((e) => e.kind === 'mission.status' && e.payload.status === 'done'), 5000, 'final event');301 await w.stop();302 await sub.close();303 const persisted = got.filter((e) => e.id > 0);304 expect(persisted[0]?.payload).toEqual({ n: 'before' });305 expect(persisted.map((e) => e.id)).toEqual([...persisted.map((e) => e.id)].sort((a, b) => a - b));306 expect(new Set(persisted.map((e) => e.id)).size).toBe(persisted.length);307 const dbIds = (await rows<{ id: string }>(db, 'SELECT id FROM events WHERE mission_id = $1 AND id >= $2 ORDER BY id', [missionId, before])).map((r) => Number(r.id));308 expect(persisted.map((e) => e.id)).toEqual(dbIds);309 const deltas = got.filter((e) => e.id === -1);310 expect(deltas.every((e) => e.kind === 'model.delta' && e.runId === rootRunId)).toBe(true);311 const text = deltas.filter((e) => e.payload.type === 'text').map((e) => e.payload.text as string);312 expect(text[0]).toBe('small');313 expect(text.slice(1).join('')).toBe(big);314 expect(text.length).toBeGreaterThan(5);315 expect(deltas.some((e) => e.payload.type === 'blob')).toBe(false);316 expect(await rows(db, `SELECT 1 FROM events WHERE kind = 'model.delta'`)).toHaveLength(0); // never persisted317 });318319 it('poll fallback delivers events even without notifications', async () => {320 const { missionId } = await newMission(db);321 const got: number[] = [];322 const sub = listenEvents((ev) => void (ev.missionId === missionId && got.push(ev.id)), { kernel: db.k, pollMs: 100 });323 await sub.ready;324 // Insert behind the kernel's back (no NOTIFY).325 const r = await rows<{ id: string }>(db, `INSERT INTO events (mission_id, kind, payload) VALUES ($1, 'system.notice', '{}') RETURNING id`, [missionId]);326 await waitFor(async () => got.includes(Number(r[0]?.id)), 3000, 'polled event');327 await sub.close();328 });329});330331describe('artifacts and migrations', () => {332 it('stores content-addressed artifacts atomically and links them to the run once', async () => {333 const { rootRunId } = await newMission(db);334 const a = await putArtifact('hello artifact', { name: 'out.txt', mime: 'text/plain', runId: rootRunId, stepSeq: 0 }, db.k);335 const b = await putArtifact(Buffer.from('hello artifact'), { name: 'out.txt', runId: rootRunId, stepSeq: 0 }, db.k);336 expect(a.sha256).toBe(createHash('sha256').update('hello artifact').digest('hex'));337 expect(a).toEqual(b);338 expect(a.size).toBe(14);339 const o = await openArtifact(a.sha256, db.k);340 expect(o?.mime).toBe('text/plain');341 expect(o?.path.endsWith(`artifacts/${a.sha256.slice(0, 2)}/${a.sha256}`)).toBe(true);342 const chunks: Buffer[] = [];343 for await (const c of o!.stream()) chunks.push(c as Buffer);344 expect(Buffer.concat(chunks).toString()).toBe('hello artifact');345 expect(await rows(db, `SELECT 1 FROM events WHERE run_id = $1 AND kind = 'artifact.created'`, [rootRunId])).toHaveLength(1);346 expect(await openArtifact('../../etc/passwd', db.k)).toBeNull();347 expect(await openArtifact('0'.repeat(64), db.k)).toBeNull();348 });349350 it('migrate is idempotent and safe to run concurrently', async () => {351 const [a, b] = await Promise.all([migrate(db.url), migrate(db.url)]);352 expect(a.applied).toEqual([]);353 expect(b.applied).toEqual([]);354 expect(a.skipped).toContain('010_kernel.sql');355 expect(a.skipped).toContain('011_workers.sql');356 });357});358