Repository Public
1/**2 * Missions, runs, steps, steering messages and artifacts. Reads are plain SQL over the kernel's3 * tables; every write the kernel owns (create mission + root run, cancel, owner message, artifact4 * content) goes through `ctx.kernel`.5 */6import path from 'node:path';7import type { FastifyInstance } from 'fastify';8import { z } from 'zod';9import {10 Budget, CreateMission, DoneCriterion, Effort, Id, MissionDto, Provider as ProviderSchema, ResultEnvelope, RunDto, RunStatus, StepDto, Usage, isTerminal,11 type Provider,12} from '@aos/shared';13import type { Queryable } from '../../db.js';14import { stepResultForBrowser } from '../../browser-view.js';15import { ArtifactDto, SteeringMessage } from '../../dto.js';16import { HttpError, authOf, notFound, parse, type Ctx } from '../../http.js';17import { ownedProject, workspaceOf } from '../projects/index.js';18import { getSettings } from '../settings/store.js';19import { todaySpendUsd } from '../spend/index.js';20import { STEP_BODY_MAX_CHARS, truncateBody } from './truncate.js';2122const IdParams = z.object({ id: Id });23const ListQuery = z.object({24 status: RunStatus.optional(),25 projectId: Id.optional(),26 limit: z.coerce.number().int().min(1).max(500).default(100),27});28const StepsQuery = z.object({29 /** Exclusive: steps with `seq > after`. */30 after: z.coerce.number().int().min(-1).max(2_000_000_000).default(-1),31 limit: z.coerce.number().int().min(1).max(500).default(200),32});33const ArtifactParams = z.object({ sha256: z.string().regex(/^[0-9a-f]{64}$/) });34const ArtifactQuery = z.object({ name: z.string().min(1).max(300).optional() });35/** `StepDto` + the truncation flag of CONTRACT §8. */36const StepOut = StepDto.extend({ truncated: z.boolean() });3738interface MissionRow {39 id: string; project_id: string | null; project_name: string | null; title: string; objective: string; done_criteria: unknown;40 status: string; budget_usd: number; spent_usd: number; network: string; root_run_id: string | null; runs: number; pending_waits: number;41 provider: string | null; model: string | null; effort: string | null;42 created_at: Date; started_at: Date | null; ended_at: Date | null;43}44const MISSION_SELECT = `SELECT m.id, m.project_id, p.name AS project_name, m.title, m.objective, m.done_criteria, m.status,45 m.budget_usd::float8 AS budget_usd, m.spent_usd::float8 AS spent_usd, m.network, m.root_run_id,46 (SELECT count(*)::int FROM runs r WHERE r.mission_id = m.id) AS runs,47 (SELECT count(*)::int FROM waits w WHERE w.mission_id = m.id AND w.status = 'pending' AND w.kind IN ('approval','question')) AS pending_waits,48 rr.provider, rr.model, rr.effort,49 m.created_at, m.started_at, m.ended_at50 FROM missions m LEFT JOIN projects p ON p.id = m.project_id LEFT JOIN runs rr ON rr.id = m.root_run_id`;51const iso = (d: Date | null): string | null => d?.toISOString() ?? null;52const missionDto = (r: MissionRow): MissionDto => {53 const criteria = z.array(DoneCriterion).safeParse(r.done_criteria);54 const provider = ProviderSchema.safeParse(r.provider);55 const effort = Effort.safeParse(r.effort);56 return {57 id: r.id, projectId: r.project_id, projectName: r.project_name, title: r.title, objective: r.objective,58 doneCriteria: criteria.success ? criteria.data : [], status: r.status as RunStatus, budgetUsd: r.budget_usd, spentUsd: r.spent_usd,59 network: r.network as 'none' | 'open', rootRunId: r.root_run_id, runs: r.runs, pendingWaits: r.pending_waits,60 createdAt: r.created_at.toISOString(), startedAt: iso(r.started_at), endedAt: iso(r.ended_at),61 // Of the root run (additive to CONTRACT §8): lists need no extra request per mission.62 provider: provider.success ? provider.data : null, model: r.model, effort: effort.success ? effort.data : null,63 };64};65export const ownedMission = (db: Queryable, id: string, userId: string): Promise<MissionRow | null> =>66 db.one<MissionRow>(`${MISSION_SELECT} WHERE m.id = $1 AND m.owner_id = $2`, [id, userId]);6768interface RunRow {69 id: string; mission_id: string; parent_run_id: string | null; depth: number; role: string; contract: { objective?: unknown } | null;70 provider: string; model: string; effort: string | null; status: string; usage: unknown; budget: unknown; sandbox_tier: number; result: unknown;71 created_at: Date; started_at: Date | null; ended_at: Date | null; pending_waits: number;72}73// Lease, fencing and workspace columns are internal: never selected for the browser.74const RUN_SELECT = `SELECT r.id, r.mission_id, r.parent_run_id, r.depth, r.role, r.contract, r.provider, r.model, r.effort, r.status, r.usage, r.budget,75 r.sandbox_tier, r.result, r.created_at, r.started_at, r.ended_at,76 (SELECT count(*)::int FROM waits w WHERE w.run_id = r.id AND w.status = 'pending' AND w.kind IN ('approval','question')) AS pending_waits77 FROM runs r JOIN missions m ON m.id = r.mission_id`;78const runDto = (r: RunRow): RunDto => {79 const usage = Usage.safeParse(r.usage ?? {});80 const budget = Budget.safeParse(r.budget);81 const result = ResultEnvelope.safeParse(r.result);82 const effort = Effort.safeParse(r.effort);83 return {84 id: r.id, missionId: r.mission_id, parentRunId: r.parent_run_id, depth: r.depth, role: r.role,85 objective: typeof r.contract?.objective === 'string' ? r.contract.objective : '',86 provider: r.provider as Provider, model: r.model, effort: effort.success ? effort.data : null, status: r.status as RunStatus,87 usage: usage.success ? usage.data : Usage.parse({}),88 budget: budget.success ? budget.data : (r.budget as Budget),89 sandboxTier: r.sandbox_tier, result: r.result != null && result.success ? result.data : null,90 createdAt: r.created_at.toISOString(), startedAt: iso(r.started_at), endedAt: iso(r.ended_at), pendingWaits: r.pending_waits,91 };92};93const ownedRun = (db: Queryable, id: string, userId: string): Promise<RunRow | null> =>94 db.one<RunRow>(`${RUN_SELECT} WHERE r.id = $1 AND m.owner_id = $2`, [id, userId]);9596/** ASCII fallback + RFC 5987 form; never a path, a quote or a control character. */97export function contentDisposition(name: string): string {98 // eslint-disable-next-line no-control-regex99 const base = (path.basename(name.replace(/\\/g, '/')) || 'artifact').replace(/[\x00-\x1f\x7f]/g, '').slice(0, 150) || 'artifact';100 const ascii = base.replace(/[^A-Za-z0-9._-]/g, '_').replace(/^\.+/, '_');101 return `attachment; filename="${ascii}"; filename*=UTF-8''${encodeURIComponent(base).replace(/['()*]/g, (c) => `%${c.charCodeAt(0).toString(16).toUpperCase()}`)}`;102}103104export default async function missionsPlugin(app: FastifyInstance, ctx: Ctx): Promise<void> {105 const { db, config, kernel } = ctx;106 const scrub = (s: string): string => config.secrets.scrub(s);107108 app.get('/missions', { config: { doc: { summary: 'Missions, newest first', query: ListQuery, response: z.array(MissionDto), responseName: 'MissionList' } } }, async (req) => {109 const a = authOf(req);110 const q = parse(ListQuery, req.query);111 const rows = await db.q<MissionRow>(112 `${MISSION_SELECT} WHERE m.owner_id = $1 AND ($2::text IS NULL OR m.status = $2) AND ($3::uuid IS NULL OR m.project_id = $3::uuid)113 ORDER BY m.created_at DESC, m.id LIMIT $4`,114 [a.userId, q.status ?? null, q.projectId ?? null, q.limit],115 );116 return rows.map(missionDto);117 });118119 app.post('/missions', { config: { doc: { summary: 'Create a mission and queue its root run', body: CreateMission, bodyName: 'CreateMission', response: MissionDto, responseName: 'Mission' } } }, async (req, reply) => {120 const a = authOf(req);121 const body = parse(CreateMission, req.body);122 const project = await ownedProject(db, body.projectId, a.userId);123 if (!project) throw notFound('project');124 const settings = await getSettings(db, a.userId);125126 // Daily cap (all missions): refuse before anything is queued.127 const todayUsd = await todaySpendUsd(db);128 if (todayUsd >= settings.dailyBudgetUsd) {129 await ctx.audit.req(req, 'mission.create', { ok: false, detail: { reason: 'daily_cap_reached' } });130 throw new HttpError(409, 'daily_cap_reached', "today's spend has reached the daily budget", { todayUsd, dailyBudgetUsd: settings.dailyBudgetUsd });131 }132133 // Provider / model / effort: explicit values, else the owner's defaults. Never substituted silently.134 const known = ctx.providers?.catalogue() ?? [];135 const { provider, model } = ((): { provider: Provider; model: string } => {136 if (!body.model && (!body.provider || body.provider === settings.defaultProvider)) return { provider: settings.defaultProvider, model: settings.defaultModel };137 if (!body.model) {138 const main = known.find((m) => m.provider === body.provider && m.tier === 'main');139 if (!main || !body.provider) throw new HttpError(400, 'model_required', 'a model is required for this provider');140 return { provider: body.provider, model: main.id };141 }142 if (body.provider) return { provider: body.provider, model: body.model };143 const hits = known.filter((m) => m.id === body.model);144 const inferred = hits.length === 1 ? hits[0]?.provider : body.model === settings.defaultModel ? settings.defaultProvider : undefined;145 if (!inferred) throw new HttpError(400, 'unknown_model', 'unknown model: give the provider explicitly');146 return { provider: inferred, model: body.model };147 })();148 const info = known.find((m) => m.provider === provider && m.id === model);149 if (known.length > 0 && !info) throw new HttpError(400, 'unknown_model', 'this model is not in the catalogue of the provider');150 if (!config.providerKeys[provider]) throw new HttpError(409, 'provider_not_configured', `no API key is configured for ${provider}`);151 let effort: Effort | null = body.effort ?? settings.defaultEffort;152 if (info && !info.efforts.includes(effort)) {153 if (body.effort) throw new HttpError(400, 'unsupported_effort', 'this model does not support the requested effort');154 effort = null; // the default does not apply to this model: provider default155 }156157 const workspaceAbs = workspaceOf(ctx, project.rel_path);158 if (!workspaceAbs) throw new HttpError(409, 'workspace_missing', 'the project directory does not exist under the workspace root');159160 const budgetUsd = body.budgetUsd ?? settings.missionBudgetUsd;161 const { missionId } = await kernel.createMission({162 ownerId: a.userId,163 projectId: project.id,164 title: (body.title ?? body.objective.replace(/\s+/g, ' ').trim().slice(0, 80)) || 'Mission',165 objective: body.objective,166 // May be empty: the kernel stores it as given and the agent then proposes criteria first (kernel-port.ts).167 doneCriteria: body.doneCriteria,168 provider,169 model,170 effort,171 budgetUsd,172 network: body.network,173 workspaceAbs,174 });175 await ctx.audit.req(req, 'mission.create', { target: missionId, detail: { projectId: project.id, provider, model, effort, budgetUsd, network: body.network } });176 const row = await ownedMission(db, missionId, a.userId);177 if (!row) throw new Error('mission not found after creation');178 return reply.code(201).send(missionDto(row));179 });180181 app.get('/missions/:id', { config: { doc: { summary: 'One mission', response: MissionDto, responseName: 'Mission' } } }, async (req) => {182 const row = await ownedMission(db, parse(IdParams, req.params).id, authOf(req).userId);183 if (!row) throw notFound('mission');184 return missionDto(row);185 });186187 app.post('/missions/:id/cancel', { config: { doc: { summary: 'Request cancellation (cascades to every run; idempotent)', response: MissionDto, responseName: 'Mission' } } }, async (req) => {188 const a = authOf(req);189 const { id } = parse(IdParams, req.params);190 const row = await ownedMission(db, id, a.userId);191 if (!row) throw notFound('mission');192 if (isTerminal(row.status as RunStatus)) return missionDto(row);193 await kernel.requestCancel(id);194 await ctx.audit.req(req, 'mission.cancel', { target: id });195 const after = await ownedMission(db, id, a.userId);196 return missionDto(after ?? row);197 });198199 app.get('/missions/:id/runs', { config: { doc: { summary: 'Runs of a mission (tree order: creation)', response: z.array(RunDto), responseName: 'RunList' } } }, async (req) => {200 const a = authOf(req);201 const { id } = parse(IdParams, req.params);202 if (!(await ownedMission(db, id, a.userId))) throw notFound('mission');203 const rows = await db.q<RunRow>(`${RUN_SELECT} WHERE r.mission_id = $1 AND m.owner_id = $2 ORDER BY r.depth, r.created_at, r.id LIMIT 2000`, [id, a.userId]);204 return rows.map(runDto);205 });206207 app.get('/runs/:id', { config: { doc: { summary: 'One run', response: RunDto, responseName: 'Run' } } }, async (req) => {208 const row = await ownedRun(db, parse(IdParams, req.params).id, authOf(req).userId);209 if (!row) throw notFound('run');210 return runDto(row);211 });212213 app.get('/runs/:id/steps', { config: { doc: { summary: 'Journal of a run (seq > after); request/result bodies truncated to 20 000 chars each', query: StepsQuery, response: z.array(StepOut), responseName: 'StepList' } } }, async (req) => {214 const a = authOf(req);215 const { id } = parse(IdParams, req.params);216 const q = parse(StepsQuery, req.query);217 if (!(await ownedRun(db, id, a.userId))) throw notFound('run');218 // idem_key and request_hash are kernel internals: not selected.219 const rows = await db.q<{ seq: number; kind: string; name: string; effect: string; status: string; request: unknown; result: unknown; error: unknown; usage: unknown; started_at: Date; completed_at: Date | null }>(220 'SELECT seq, kind, name, effect, status, request, result, error, usage, started_at, completed_at FROM steps WHERE run_id = $1 AND seq > $2 ORDER BY seq LIMIT $3',221 [id, q.after, q.limit],222 );223 return rows.map((s): z.infer<typeof StepOut> => {224 const request = truncateBody(s.request, STEP_BODY_MAX_CHARS, scrub);225 const result = truncateBody(stepResultForBrowser(s.kind, s.result), STEP_BODY_MAX_CHARS, scrub); // never the provider-native transcript226 const error = truncateBody(s.error, STEP_BODY_MAX_CHARS, scrub);227 const usage = Usage.partial().safeParse(s.usage);228 return {229 runId: id, seq: s.seq, kind: s.kind, name: s.name, effect: s.effect, status: s.status as StepDto['status'],230 request: request.value, result: result.value ?? null, error: error.value ?? null,231 usage: s.usage != null && usage.success ? usage.data : null,232 startedAt: s.started_at.toISOString(), completedAt: iso(s.completed_at),233 truncated: request.truncated || result.truncated || error.truncated,234 };235 });236 });237238 app.post('/runs/:id/message', { config: { doc: { summary: 'Steering message from the owner to a live run', body: SteeringMessage, bodyName: 'SteeringMessage' } } }, async (req) => {239 const a = authOf(req);240 const { id } = parse(IdParams, req.params);241 const body = parse(SteeringMessage, req.body);242 const run = await ownedRun(db, id, a.userId);243 if (!run) throw notFound('run');244 const ended = (): HttpError => new HttpError(409, 'run_ended', 'this run has ended');245 if (isTerminal(run.status as RunStatus)) throw ended();246 try {247 await kernel.postOwnerMessage(id, body.text);248 } catch (err) {249 // The run ended between the check and the kernel call.250 const now = await ownedRun(db, id, a.userId);251 if (!now || isTerminal(now.status as RunStatus)) throw ended();252 throw err;253 }254 await ctx.audit.req(req, 'run.message', { target: id, detail: { chars: body.text.length } }); // never the text255 return { ok: true };256 });257258 app.get('/runs/:id/artifacts', { config: { doc: { summary: 'Artifacts produced by a run', response: z.array(ArtifactDto), responseName: 'ArtifactList' } } }, async (req) => {259 const a = authOf(req);260 const { id } = parse(IdParams, req.params);261 if (!(await ownedRun(db, id, a.userId))) throw notFound('run');262 const rows = await db.q<{ sha256: string; name: string; size: number; mime: string; step_seq: number | null; created_at: Date }>(263 `SELECT ra.sha256, ra.name, ar.size, ar.mime, ra.step_seq, ra.created_at264 FROM run_artifacts ra JOIN artifacts ar ON ar.sha256 = ra.sha256 WHERE ra.run_id = $1 ORDER BY ra.created_at, ra.name LIMIT 2000`,265 [id],266 );267 return rows.map((r): ArtifactDto => ({ sha256: r.sha256, name: r.name, size: r.size, mime: r.mime, stepSeq: r.step_seq, createdAt: r.created_at.toISOString() }));268 });269270 app.get('/artifacts/:sha256', { config: { doc: { summary: 'Download an artifact (always an attachment, never rendered)', query: ArtifactQuery } } }, async (req, reply) => {271 const a = authOf(req);272 const { sha256 } = parse(ArtifactParams, req.params);273 const q = parse(ArtifactQuery, req.query);274 // Reachable only through a run of a mission the caller owns; the display name must be one the journal recorded.275 const rows = await db.q<{ name: string; size: number }>(276 `SELECT ra.name, ar.size FROM run_artifacts ra JOIN artifacts ar ON ar.sha256 = ra.sha256277 JOIN runs r ON r.id = ra.run_id JOIN missions m ON m.id = r.mission_id278 WHERE ra.sha256 = $1 AND m.owner_id = $2 ORDER BY ra.created_at LIMIT 200`,279 [sha256, a.userId],280 );281 const hit = q.name ? rows.find((r) => r.name === q.name) : rows[0];282 if (!hit) throw notFound('artifact');283 const stream = await kernel.openArtifact(sha256);284 if (!stream) throw new HttpError(404, 'artifact_missing', 'the artifact content is no longer stored');285 return reply286 .header('Content-Type', 'application/octet-stream') // the recorded mime is never trusted for rendering287 .header('Content-Disposition', contentDisposition(hit.name))288 .header('X-Content-Type-Options', 'nosniff')289 .header('Content-Security-Policy', "default-src 'none'; sandbox")290 .header('Content-Length', hit.size)291 .send(stream);292 });293}294