Repository Public
1/**2 * SQL-backed stand-in for `@aos/kernel` (tests only). It writes the rows CONTRACT §2 documents3 * for the operations the server delegates, so the server's SQL reads are exercised against the4 * real schema. It executes nothing: no leases, no steps, no agent. The `seed*` / `emit*` methods5 * are test fixtures that play the part of a worker.6 */7import { createHash } from 'node:crypto';8import fs from 'node:fs';9import path from 'node:path';10import type { Readable } from 'node:stream';11import { Budget, ENGINE_VERSION, type AosEvent, type EventKind, type Trust } from '@aos/shared';12import type { Db, Queryable } from '../db.js';13import type { CreateMissionInput, KernelPort } from '../kernel-port.js';1415interface EventRow { id: number; mission_id: string; run_id: string | null; seq: number | null; ts: Date; kind: string; trust: string; payload: Record<string, unknown> }1617export class FakeKernel implements KernelPort {18 private readonly listeners = new Set<(ev: AosEvent) => void>();19 private timer: NodeJS.Timeout | null = null;20 private cursor = 0;21 private polling = false;22 calls = { createMission: 0, requestCancel: 0, resolveWait: 0, postOwnerMessage: 0, openArtifact: 0 };2324 constructor(25 private readonly db: Db,26 private readonly dataDir: string,27 ) {}2829 private async event(tx: Queryable, missionId: string, runId: string | null, kind: EventKind, payload: Record<string, unknown>, trust: Trust = 'system', seq: number | null = null): Promise<number> {30 const row = await tx.one<{ id: number }>(31 'INSERT INTO events (mission_id, run_id, seq, kind, trust, payload) VALUES ($1, $2, $3, $4, $5, $6) RETURNING id',32 [missionId, runId, seq, kind, trust, JSON.stringify(payload)],33 );34 return row?.id ?? -1;35 }3637 // ---------------- KernelPort ----------------38 async createMission(input: CreateMissionInput): Promise<{ missionId: string; rootRunId: string }> {39 this.calls.createMission++;40 // Same root contract as the kernel builds: criteria as given, possibly empty.41 const contract = { objective: input.objective, contextRefs: [], boundaries: '', outputSchema: null, tools: [], doneCriteria: input.doneCriteria, contextMode: 'fresh' };42 return this.db.withTx(async (tx) => {43 const m = await tx.one<{ id: string }>(44 `INSERT INTO missions (owner_id, project_id, title, objective, done_criteria, budget_usd, network)45 VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING id`,46 [input.ownerId, input.projectId, input.title, input.objective, JSON.stringify(input.doneCriteria.map((text, n) => ({ id: `c${n + 1}`, text, passes: false }))), input.budgetUsd, input.network],47 );48 if (!m) throw new Error('mission insert failed');49 const r = await tx.one<{ id: string }>(50 `INSERT INTO runs (mission_id, contract, provider, model, effort, workspace_abs, budget, engine_version)51 VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING id`,52 [m.id, JSON.stringify(contract), input.provider, input.model, input.effort, input.workspaceAbs, JSON.stringify(Budget.parse({ usd: input.budgetUsd })), ENGINE_VERSION],53 );54 if (!r) throw new Error('run insert failed');55 await tx.q('UPDATE missions SET root_run_id = $2 WHERE id = $1', [m.id, r.id]);56 await this.event(tx, m.id, null, 'mission.created', { title: input.title, projectId: input.projectId, rootRunId: r.id, status: 'queued' });57 await this.event(tx, m.id, r.id, 'run.created', { parentRunId: null, role: 'agent', provider: input.provider, model: input.model, status: 'queued' });58 return { missionId: m.id, rootRunId: r.id };59 });60 }6162 async requestCancel(missionId: string): Promise<void> {63 this.calls.requestCancel++;64 await this.db.withTx(async (tx) => {65 await tx.q('UPDATE runs SET cancel_requested = true WHERE mission_id = $1', [missionId]);66 await tx.q("UPDATE waits SET status = 'cancelled', resolved_at = now() WHERE mission_id = $1 AND status = 'pending'", [missionId]);67 const ended = await tx.q<{ id: string }>(68 "UPDATE runs SET status = 'cancelled', ended_at = now() WHERE mission_id = $1 AND status IN ('queued','waiting') RETURNING id",69 [missionId],70 );71 for (const r of ended) await this.event(tx, missionId, r.id, 'run.status', { status: 'cancelled' });72 const m = await tx.q(73 `UPDATE missions SET status = 'cancelled', ended_at = now()74 WHERE id = $1 AND status IN ('queued','waiting') RETURNING id`,75 [missionId],76 );77 if (m.length) await this.event(tx, missionId, null, 'mission.status', { status: 'cancelled' });78 });79 }8081 async resolveWait(waitId: string, resolution: Record<string, unknown>, actor: string): Promise<{ resolved: boolean }> {82 this.calls.resolveWait++;83 return this.db.withTx(async (tx) => {84 if (!(await tx.one('SELECT 1 AS x FROM waits WHERE id = $1', [waitId]))) throw new Error(`wait ${waitId} not found`);85 const w = await tx.one<{ run_id: string; mission_id: string; kind: string }>(86 "UPDATE waits SET status = 'resolved', resolution = $2, resolved_at = now() WHERE id = $1 AND status = 'pending' RETURNING run_id, mission_id, kind",87 [waitId, JSON.stringify(resolution)],88 );89 if (!w) return { resolved: false };90 await tx.q("UPDATE runs SET status = 'queued' WHERE id = $1 AND status = 'waiting'", [w.run_id]);91 await this.event(tx, w.mission_id, w.run_id, 'wait.resolved', { waitId, kind: w.kind, status: 'resolved', resolution, actor }, 'owner');92 return { resolved: true };93 });94 }9596 async postOwnerMessage(runId: string, text: string): Promise<void> {97 this.calls.postOwnerMessage++;98 const r = await this.db.one<{ mission_id: string; status: string }>('SELECT mission_id, status FROM runs WHERE id = $1', [runId]);99 if (!r) throw new Error('run not found');100 if (!['queued', 'running', 'waiting'].includes(r.status)) throw new Error(`run is ${r.status}`);101 await this.event(this.db, r.mission_id, runId, 'owner.message', { text }, 'owner');102 }103104 async listenEvents(cb: (event: AosEvent) => void): Promise<() => Promise<void>> {105 if (this.listeners.size === 0) {106 this.cursor = (await this.db.one<{ id: number | null }>('SELECT max(id) AS id FROM events'))?.id ?? 0;107 this.timer = setInterval(() => void this.poll(), 15);108 }109 this.listeners.add(cb);110 return async () => {111 this.listeners.delete(cb);112 if (this.listeners.size === 0 && this.timer) {113 clearInterval(this.timer);114 this.timer = null;115 }116 };117 }118119 async openArtifact(sha256: string): Promise<Readable | null> {120 this.calls.openArtifact++;121 const file = this.artifactPath(sha256);122 if (!fs.existsSync(file)) return null;123 return fs.createReadStream(file);124 }125126 // ---------------- fixtures (the part a worker would play) ----------------127 private async poll(): Promise<void> {128 if (this.polling) return;129 this.polling = true;130 try {131 const rows = await this.db.q<EventRow>('SELECT id, mission_id, run_id, seq, ts, kind, trust, payload FROM events WHERE id > $1 ORDER BY id LIMIT 1000', [this.cursor]);132 for (const r of rows) {133 this.cursor = r.id;134 const ev: AosEvent = { id: r.id, missionId: r.mission_id, runId: r.run_id, seq: r.seq, ts: r.ts.toISOString(), kind: r.kind as EventKind, trust: r.trust as Trust, payload: r.payload };135 for (const l of this.listeners) l(ev);136 }137 } catch {138 /* pool ended during shutdown */139 } finally {140 this.polling = false;141 }142 }143 /** Resolves once every event stored so far has been handed to the listeners. */144 async drained(): Promise<void> {145 const head = (await this.db.one<{ id: number | null }>('SELECT max(id) AS id FROM events'))?.id ?? 0;146 while (this.listeners.size > 0 && this.cursor < head) {147 await this.poll();148 await new Promise((r) => setTimeout(r, 2));149 }150 }151 /** Appends a persisted event (delivered to listeners by the poll). */152 emit(missionId: string, runId: string | null, kind: EventKind, payload: Record<string, unknown>, trust: Trust = 'system', seq: number | null = null): Promise<number> {153 return this.event(this.db, missionId, runId, kind, payload, trust, seq);154 }155 /** Transient streaming delta: pushed to listeners, never stored. */156 emitTransient(missionId: string, runId: string, payload: Record<string, unknown>): void {157 const ev: AosEvent = { id: -1, missionId, runId, seq: null, ts: new Date().toISOString(), kind: 'model.delta', trust: 'agent', payload };158 for (const l of this.listeners) l(ev);159 }160 artifactPath(sha256: string): string {161 return path.join(this.dataDir, 'artifacts', sha256.slice(0, 2), sha256);162 }163 async seedArtifact(runId: string, name: string, data: Buffer | string, mime = 'application/octet-stream', stepSeq: number | null = null): Promise<string> {164 const bytes = Buffer.isBuffer(data) ? data : Buffer.from(data);165 const sha256 = createHash('sha256').update(bytes).digest('hex');166 const file = this.artifactPath(sha256);167 fs.mkdirSync(path.dirname(file), { recursive: true });168 fs.writeFileSync(file, bytes);169 await this.db.q('INSERT INTO artifacts (sha256, size, mime) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING', [sha256, bytes.length, mime]);170 await this.db.q('INSERT INTO run_artifacts (run_id, sha256, name, step_seq) VALUES ($1, $2, $3, $4)', [runId, sha256, name, stepSeq]);171 return sha256;172 }173 async seedStep(runId: string, seq: number, o: { kind?: string; name?: string; effect?: string; status?: string; request?: unknown; result?: unknown; error?: unknown; usage?: unknown }): Promise<void> {174 const completed = (o.status ?? 'completed') !== 'started';175 await this.db.q(176 `INSERT INTO steps (run_id, seq, kind, name, effect, status, request_hash, request, result, error, usage, idem_key, completed_at)177 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, CASE WHEN $13::boolean THEN now() END)`,178 [179 runId, seq, o.kind ?? 'tool_call', o.name ?? 'read_file', o.effect ?? 'read', o.status ?? 'completed', 'HASH-INTERNAL',180 JSON.stringify(o.request ?? {}), o.result === undefined ? null : JSON.stringify(o.result), o.error === undefined ? null : JSON.stringify(o.error),181 o.usage === undefined ? null : JSON.stringify(o.usage), `IDEM-INTERNAL-${seq}`, completed,182 ],183 );184 }185 async seedWait(runId: string, stepSeq: number, kind: 'approval' | 'question' | 'timer' | 'child', payload: Record<string, unknown>, resolveBy: Date | null = null): Promise<string> {186 return this.db.withTx(async (tx) => {187 const r = await tx.one<{ mission_id: string }>("UPDATE runs SET status = 'waiting' WHERE id = $1 RETURNING mission_id", [runId]);188 if (!r) throw new Error('run not found');189 const w = await tx.one<{ id: string }>(190 'INSERT INTO waits (run_id, step_seq, mission_id, kind, payload, resolve_by) VALUES ($1, $2, $3, $4, $5, $6) RETURNING id',191 [runId, stepSeq, r.mission_id, kind, JSON.stringify(payload), resolveBy],192 );193 if (!w) throw new Error('wait insert failed');194 await tx.q("UPDATE missions SET status = 'waiting' WHERE id = $1", [r.mission_id]);195 await this.event(tx, r.mission_id, runId, 'wait.created', { waitId: w.id, kind, payload, resolveBy: resolveBy?.toISOString() ?? null }, 'system', stepSeq);196 return w.id;197 });198 }199 stop(): void {200 if (this.timer) clearInterval(this.timer);201 this.timer = null;202 this.listeners.clear();203 }204}205