Repository Public
1/** Mission/run creation, cancellation cascade, owner steering messages. */2import { z } from 'zod';3import { Budget, DelegationContract, Effort, ENGINE_VERSION, isTerminal, Provider, type RunStatus } from '@aos/shared';4import { canonicalJson } from './canonical.js';5import { getKernel, withTx, type Kernel } from './db.js';6import { KernelError } from './errors.js';7import { announceRunStatus, cancelPendingWaits, envelope, finalizeRun } from './runs.js';89export const CreateMissionInput = z.object({10 ownerId: z.string().uuid(),11 projectId: z.string().uuid().nullable().default(null),12 title: z.string().min(1).max(160).optional(), // default: first 80 chars of the objective13 objective: z.string().min(1).max(8000),14 /** May be empty: the agent then proposes criteria first (CONTRACT §5.1). */15 doneCriteria: z.array(z.string().min(1).max(1000)).max(50).default([]),16 provider: Provider,17 model: z.string().min(1).max(80),18 effort: Effort.nullable().default(null),19 budgetUsd: z.number().positive(),20 /** Run limits other than usd (turns, wallSec, tokens); defaults from shared `Budget`. */21 limits: Budget.omit({ usd: true }).partial().default({}),22 network: z.enum(['none', 'open']).default('none'),23 workspaceAbs: z.string().min(1),24 role: z.string().min(1).max(40).default('agent'),25 sandboxTier: z.number().int().min(0).max(3).default(1),26 boundaries: z.string().max(4000).default(''),27 contextRefs: z.array(z.string().max(500)).max(50).default([]),28 tools: z.array(z.string()).default([]),29});30export type CreateMissionInput = z.input<typeof CreateMissionInput>;3132/** Creates the mission and its root run (queued) in one transaction. */33export async function createMission(input: CreateMissionInput, k: Kernel = getKernel()): Promise<{ missionId: string; rootRunId: string }> {34 const i = CreateMissionInput.parse(input);35 const budget = Budget.parse({ usd: i.budgetUsd, ...i.limits });36 // The root contract may have no done criteria yet (DelegationContract.min(1) applies to spawns).37 const contract: DelegationContract = {38 objective: i.objective,39 contextRefs: i.contextRefs,40 boundaries: i.boundaries,41 outputSchema: null,42 tools: i.tools,43 doneCriteria: i.doneCriteria,44 contextMode: 'fresh',45 };46 const criteria = i.doneCriteria.map((text, n) => ({ id: `c${n + 1}`, text, passes: false }));47 const title = i.title ?? i.objective.slice(0, 80);48 return withTx(k, async (tx) => {49 const m = await tx.one<{ id: string }>(50 `INSERT INTO missions (owner_id, project_id, title, objective, done_criteria, budget_usd, network)51 VALUES ($1,$2,$3,$4,$5::jsonb,$6,$7) RETURNING id`,52 [i.ownerId, i.projectId, title, i.objective, canonicalJson(criteria), i.budgetUsd, i.network],53 );54 const missionId = (m as { id: string }).id;55 const r = await tx.one<{ id: string }>(56 `INSERT INTO runs (mission_id, parent_run_id, depth, role, contract, provider, model, effort, workspace_abs, sandbox_tier, budget, engine_version)57 VALUES ($1, NULL, 0, $2, $3::jsonb, $4, $5, $6, $7, $8, $9::jsonb, $10) RETURNING id`,58 [missionId, i.role, canonicalJson(contract), i.provider, i.model, i.effort, i.workspaceAbs, i.sandboxTier, canonicalJson(budget), ENGINE_VERSION],59 );60 const rootRunId = (r as { id: string }).id;61 await tx.q('UPDATE missions SET root_run_id = $2 WHERE id = $1', [missionId, rootRunId]);62 tx.emit({ missionId, kind: 'mission.created', payload: { title, projectId: i.projectId, rootRunId, status: 'queued' } });63 tx.emit({ missionId, runId: rootRunId, kind: 'run.created', payload: { parentRunId: null, role: i.role, provider: i.provider, model: i.model, status: 'queued' } });64 tx.wake('q');65 return { missionId, rootRunId };66 });67}6869export const CreateRunOptions = z.object({70 provider: Provider,71 model: z.string().min(1).max(80),72 effort: Effort.nullable().default(null),73 budget: Budget,74 role: z.string().min(1).max(40).default('worker'),75 workspaceAbs: z.string().min(1).optional(), // default: the parent's76 sandboxTier: z.number().int().min(0).max(3).optional(),77});78export type CreateRunOptions = z.input<typeof CreateRunOptions>;7980/** Child run (later phases). The contract is validated strictly: a spawn without done criteria is rejected. */81export async function createRun(82 parentRunId: string,83 contract: z.input<typeof DelegationContract>,84 options: CreateRunOptions,85 k: Kernel = getKernel(),86): Promise<{ runId: string }> {87 const c = DelegationContract.parse(contract);88 const o = CreateRunOptions.parse(options);89 return withTx(k, async (tx) => {90 const p = await tx.one<{ mission_id: string; depth: number; status: RunStatus; workspace_abs: string; sandbox_tier: number; cancel_requested: boolean }>(91 'SELECT mission_id, depth, status, workspace_abs, sandbox_tier, cancel_requested FROM runs WHERE id = $1 FOR UPDATE',92 [parentRunId],93 );94 if (!p) throw new KernelError(`createRun: parent run ${parentRunId} not found`);95 if (isTerminal(p.status) || p.cancel_requested) throw new KernelError(`createRun: parent run is ${p.cancel_requested ? 'being cancelled' : p.status}`);96 const r = await tx.one<{ id: string }>(97 `INSERT INTO runs (mission_id, parent_run_id, depth, role, contract, provider, model, effort, workspace_abs, sandbox_tier, budget, engine_version)98 VALUES ($1,$2,$3,$4,$5::jsonb,$6,$7,$8,$9,$10,$11::jsonb,$12) RETURNING id`,99 [p.mission_id, parentRunId, p.depth + 1, o.role, canonicalJson(c), o.provider, o.model, o.effort, o.workspaceAbs ?? p.workspace_abs, o.sandboxTier ?? p.sandbox_tier, canonicalJson(o.budget), ENGINE_VERSION],100 );101 const runId = (r as { id: string }).id;102 tx.emit({ missionId: p.mission_id, runId, kind: 'run.created', payload: { parentRunId, role: o.role, provider: o.provider, model: o.model, status: 'queued' } });103 tx.wake('q');104 return { runId };105 });106}107108/**109 * Cancels a mission (all its runs) or a run and its descendants. `id` is a mission id or a run id.110 * queued / waiting runs, and running runs whose lease has expired, become `cancelled` immediately111 * (fencing token bumped so a zombie cannot write); live running runs get `cancel_requested` and their112 * worker finalizes them at its next step boundary or heartbeat. Pending waits are cancelled either way.113 */114export async function requestCancel(id: string, k: Kernel = getKernel()): Promise<{ cancelled: string[]; requested: string[] }> {115 z.string().uuid().parse(id);116 return withTx(k, async (tx) => {117 const runs = await tx.q<{ id: string; mission_id: string; parent_run_id: string | null; status: RunStatus; live: boolean }>(118 `WITH RECURSIVE tree AS (119 SELECT id FROM runs WHERE mission_id = $1 OR id = $1120 UNION SELECT r.id FROM runs r JOIN tree t ON r.parent_run_id = t.id)121 SELECT r.id, r.mission_id, r.parent_run_id, r.status,122 (r.status = 'running' AND r.lease_expires_at IS NOT NULL AND r.lease_expires_at >= now()) AS live123 FROM runs r WHERE r.id IN (SELECT id FROM tree) ORDER BY r.id FOR UPDATE OF r`,124 [id],125 );126 const out = { cancelled: [] as string[], requested: [] as string[] };127 for (const run of runs) {128 if (isTerminal(run.status)) continue;129 if (run.live) {130 await tx.q('UPDATE runs SET cancel_requested = true WHERE id = $1', [run.id]);131 await cancelPendingWaits(tx, run);132 tx.wake(`c:${run.id}`);133 out.requested.push(run.id);134 } else {135 await tx.q('UPDATE runs SET cancel_requested = true, fencing_token = fencing_token + 1 WHERE id = $1', [run.id]);136 await finalizeRun(tx, run, envelope('cancelled', 'Cancelled by the owner.'));137 out.cancelled.push(run.id);138 }139 }140 return out;141 });142}143144/** Steering message from the owner: an `owner.message` event the run reads through ctx.ownerMessages(). */145export async function postOwnerMessage(runId: string, text: string, k: Kernel = getKernel()): Promise<{ eventId: number }> {146 const body = z.string().min(1).max(8000).parse(text);147 return withTx(k, async (tx) => {148 const run = await tx.one<{ mission_id: string; status: RunStatus }>('SELECT mission_id, status FROM runs WHERE id = $1', [runId]);149 if (!run) throw new KernelError(`postOwnerMessage: run ${runId} not found`);150 if (isTerminal(run.status)) throw new KernelError(`postOwnerMessage: run is ${run.status}`);151 tx.emit({ missionId: run.mission_id, runId, kind: 'owner.message', trust: 'owner', payload: { text: body } });152 const ids = await tx.flush();153 return { eventId: ids[0] as number };154 });155}156157export { announceRunStatus };158