Repository Public
1/** Row types and the run/mission status helpers shared by every kernel module. */2import { isTerminal, type ResultEnvelope, type RunStatus } from '@aos/shared';3import { canonicalJson } from './canonical.js';4import type { Tx } from './db.js';56/** The columns the kernel needs from `runs`. bigint columns arrive as string or number (driver config). */7export interface RunRow {8 id: string;9 mission_id: string;10 parent_run_id: string | null;11 contract: unknown;12 status: RunStatus;13 workspace_abs: string;14 budget: { usd: number; tokens?: number; turns?: number; wallSec?: number };15 usage: Record<string, number>;16 next_seq: number;17 next_emit: number;18 fencing_token: string | number;19 cancel_requested: boolean;20 lease_owner: string | null;21 started_at: Date | null;22}23export const RUN_COLS =24 'id, mission_id, parent_run_id, contract, status, workspace_abs, budget, usage, next_seq, next_emit, fencing_token, cancel_requested, lease_owner, started_at';2526type RunRef = Pick<RunRow, 'id' | 'mission_id' | 'parent_run_id'>;2728/**29 * Emits run.status and mirrors the status onto the mission when the run is its root.30 * The caller has already updated (and therefore locked) the runs row in this transaction.31 */32export async function announceRunStatus(tx: Tx, run: RunRef, status: RunStatus, extra: Record<string, unknown> = {}): Promise<void> {33 tx.emit({ missionId: run.mission_id, runId: run.id, kind: 'run.status', payload: { status, ...extra } });34 if (run.parent_run_id !== null) return;35 const terminal = isTerminal(status);36 const m = await tx.q(37 `UPDATE missions SET status = $2,38 started_at = COALESCE(started_at, CASE WHEN $2 = 'running' THEN now() END),39 ended_at = CASE WHEN $3 THEN now() ELSE NULL END40 WHERE id = $1 AND root_run_id = $4 RETURNING id`,41 [run.mission_id, status, terminal, run.id],42 );43 if (m.length > 0) tx.emit({ missionId: run.mission_id, runId: null, kind: 'mission.status', payload: { status } });44}4546export async function cancelPendingWaits(tx: Tx, run: RunRef): Promise<void> {47 const ws = await tx.q<{ id: string; kind: string; step_seq: number }>(48 `UPDATE waits SET status = 'cancelled', resolved_at = now() WHERE run_id = $1 AND status = 'pending' RETURNING id, kind, step_seq`,49 [run.id],50 );51 for (const w of ws) {52 tx.emit({ missionId: run.mission_id, runId: run.id, seq: w.step_seq, kind: 'wait.resolved', payload: { waitId: w.id, kind: w.kind, status: 'cancelled' } });53 }54}5556/** Terminal transition: result stored, lease cleared, pending waits cancelled, events, mission mirror. */57export async function finalizeRun(tx: Tx, run: RunRef, envelope: ResultEnvelope): Promise<void> {58 await tx.q(59 `UPDATE runs SET status = $2, result = $3::jsonb, ended_at = now(), lease_owner = NULL, lease_expires_at = NULL, wake_at = NULL WHERE id = $1`,60 [run.id, envelope.status, canonicalJson(envelope)],61 );62 await cancelPendingWaits(tx, run);63 await announceRunStatus(tx, run, envelope.status, { summary: envelope.summary.slice(0, 500) });64}6566export function envelope(status: ResultEnvelope['status'], summary: string): ResultEnvelope {67 return { status, summary: summary.slice(0, 8000), payload: null, artifacts: [], evidence: [], openQuestions: [] };68}69