Repository Public
1/** Agent task endpoints. Commands are logged nowhere (audit + logs carry the task id only). */2import type { FastifyInstance } from 'fastify';3import { z } from 'zod';4import { AgentTask, CreateTask, Id, type RtTaskInfo, type RtTaskSpec } from '@cr/shared';5import { HttpError, authOf, notFound, parse, type Ctx } from '../../http.js';6import { absUnder, runtimeToHttp, validateRel } from '../../paths.js';7import { RuntimeError } from '../../runtime-client.js';8import { contentDisposition } from '../files/index.js';9import { projectDir } from '../sessions/index.js';10import { TERMINAL_TASK_STATES, type TaskRow } from './service.js';1112const IdParams = z.object({ id: Id });13const ArtifactParams = z.object({ id: Id, artifactId: Id });14const ListQuery = z.object({15 limit: z.coerce.number().int().min(1).max(500).default(100),16 projectId: Id.optional(),17});18const LogQuery = z.object({19 offset: z.coerce.number().int().min(0).default(0),20 limit: z.coerce.number().int().min(1).max(4 * 1024 * 1024).default(1024 * 1024),21});2223export default async function tasksPlugin(app: FastifyInstance, ctx: Ctx): Promise<void> {24 const { db, rt, audit, config } = ctx;25 const svc = ctx.tasks;26 const owned = async (id: string, ownerId: string): Promise<TaskRow> => {27 const row = await svc.owned(id, ownerId);28 if (!row) throw notFound('task');29 return row;30 };31 const hostAllowed = (host: string): boolean => host === 'local' || config.execHosts.includes(host);3233 /** Inserts a QUEUED row and hands it to the runtime (⇒ RUNNING, or FAILED if it cannot start). */34 async function launch(ownerId: string, t: { projectId: string | null; title: string; command: string; host: string; cwdAbs: string; cwdRel: string; timeoutSec: number | null; idempotent: boolean; artifactGlobs: string[]; retryOf: string | null }): Promise<TaskRow> {35 const row = await db.one<TaskRow>(36 `INSERT INTO agent_tasks (owner_id, project_id, title, command, host, cwd_abs, cwd_rel, state, timeout_sec, idempotent, retry_of, artifact_globs)37 VALUES ($1, $2, $3, $4, $5, $6, $7, 'QUEUED', $8, $9, $10, $11) RETURNING *`,38 [ownerId, t.projectId, t.title, t.command, t.host, t.cwdAbs, t.cwdRel, t.timeoutSec, t.idempotent, t.retryOf, t.artifactGlobs],39 );40 if (!row) throw new Error('insert failed');41 const spec: RtTaskSpec = { id: row.id, command: t.command, cwd: t.cwdAbs, host: t.host, timeoutSec: t.timeoutSec, env: {} };42 try {43 const info = await rt.json<RtTaskInfo>('POST', '/tasks', spec);44 const cur = await svc.owned(row.id, ownerId);45 if (cur?.state !== 'QUEUED') return cur ?? row; // a runtime event already advanced it46 return (await svc.transition(row.id, 'RUNNING', { startedAtMs: info?.startedAtMs ?? null })) ?? row;47 } catch (err) {48 await svc.transition(row.id, 'FAILED');49 return runtimeToHttp(err);50 }51 }5253 app.get('/tasks', { config: { doc: { summary: 'List tasks (newest first)', query: ListQuery, response: z.array(AgentTask), responseName: 'AgentTaskList' } } }, async (req) => {54 const a = authOf(req);55 const q = parse(ListQuery, req.query);56 const rows = await db.q<TaskRow>(57 'SELECT * FROM agent_tasks WHERE owner_id = $1 AND ($2::uuid IS NULL OR project_id = $2::uuid) ORDER BY created_at DESC LIMIT $3',58 [a.userId, q.projectId ?? null, q.limit],59 );60 return Promise.all(rows.map((r) => svc.dto(r)));61 });6263 app.post('/tasks', { config: { doc: { summary: 'Create and start a task', body: CreateTask, bodyName: 'CreateTask', response: AgentTask, responseName: 'AgentTask' } } }, async (req, reply) => {64 const a = authOf(req);65 const body = parse(CreateTask, req.body);66 if (!hostAllowed(body.host)) {67 await audit.req(req, 'task.create', { ok: false, detail: { reason: 'host_not_allowed', host: body.host } });68 throw new HttpError(403, 'host_not_allowed', 'this host is not on the remote execution allowlist');69 }70 const dir = await projectDir(ctx, a.userId, body.projectId);71 const rel = validateRel(body.cwd);72 let row: TaskRow;73 try {74 row = await launch(a.userId, {75 projectId: body.projectId, title: body.title, command: body.command, host: body.host, cwdAbs: absUnder(dir.abs, rel), cwdRel: rel,76 timeoutSec: body.timeoutSec, idempotent: body.idempotent, artifactGlobs: body.artifactGlobs, retryOf: null,77 });78 } catch (err) {79 await audit.req(req, 'task.create', { ok: false });80 throw err;81 }82 await audit.req(req, 'task.create', { target: row.id, detail: { host: body.host, projectId: body.projectId } });83 void reply.code(201);84 return svc.dto(row);85 });8687 app.get('/tasks/:id', { config: { doc: { summary: 'Get one task', response: AgentTask, responseName: 'AgentTask' } } }, async (req) => {88 const { id } = parse(IdParams, req.params);89 return svc.dto(await owned(id, authOf(req).userId));90 });9192 app.post('/tasks/:id/cancel', { config: { doc: { summary: 'Cancel a queued or running task', response: AgentTask, responseName: 'AgentTask' } } }, async (req) => {93 const a = authOf(req);94 const { id } = parse(IdParams, req.params);95 const row = await owned(id, a.userId);96 if (TERMINAL_TASK_STATES.includes(row.state)) throw new HttpError(409, 'task_finished', 'task already finished');97 if (row.state === 'QUEUED') {98 await svc.transition(id, 'CANCELLED');99 } else {100 await rt.json('GET', '/health', undefined, { timeoutMs: 5000 }).catch(runtimeToHttp);101 await svc.transition(id, 'CANCELLING');102 try {103 const info = await rt.json<RtTaskInfo | null>('DELETE', `/tasks/${id}`);104 if (info && typeof info === 'object' && typeof info.state === 'string' && info.state !== 'RUNNING') await svc.applyInfo({ ...info, id });105 } catch (err) {106 if (err instanceof RuntimeError && err.status === 404) await svc.transition(id, 'LOST');107 else return runtimeToHttp(err);108 }109 }110 await audit.req(req, 'task.cancel', { target: id });111 return svc.dto(await owned(id, a.userId));112 });113114 app.post('/tasks/:id/retry', { config: { doc: { summary: 'Re-run a finished task (only when idempotent)', response: AgentTask, responseName: 'AgentTask' } } }, async (req, reply) => {115 const a = authOf(req);116 const { id } = parse(IdParams, req.params);117 const row = await owned(id, a.userId);118 if (!row.idempotent) throw new HttpError(409, 'not_idempotent', 'only tasks marked idempotent can be retried');119 if (!TERMINAL_TASK_STATES.includes(row.state)) throw new HttpError(409, 'task_active', 'task has not finished');120 if (!hostAllowed(row.host)) throw new HttpError(403, 'host_not_allowed', 'this host is not on the remote execution allowlist');121 const next = await launch(a.userId, {122 projectId: row.project_id, title: row.title, command: row.command, host: row.host, cwdAbs: row.cwd_abs, cwdRel: row.cwd_rel,123 timeoutSec: row.timeout_sec, idempotent: true, artifactGlobs: row.artifact_globs, retryOf: row.id,124 });125 await audit.req(req, 'task.retry', { target: next.id, detail: { retryOf: id } });126 void reply.code(201);127 return svc.dto(next);128 });129130 app.get('/tasks/:id/log', { config: { doc: { summary: 'Task log bytes from an offset (X-Log-Size header)', query: LogQuery } } }, async (req, reply) => {131 const { id } = parse(IdParams, req.params);132 const q = parse(LogQuery, req.query);133 await owned(id, authOf(req).userId);134 const res = await rt.raw('GET', `/tasks/${id}/log`, { query: { offset: q.offset, limit: q.limit } }).catch(runtimeToHttp);135 const size = res.headers['x-log-size'];136 if (typeof size === 'string' && /^\d+$/.test(size)) void reply.header('X-Log-Size', size);137 return reply.header('Content-Type', 'application/octet-stream').header('X-Content-Type-Options', 'nosniff').send(res.body);138 });139140 app.get('/tasks/:id/artifacts/:artifactId', { config: { doc: { summary: 'Download a task artifact (attachment, streamed)' } } }, async (req, reply) => {141 const { id, artifactId } = parse(ArtifactParams, req.params);142 const task = await owned(id, authOf(req).userId);143 const art = await db.one<{ name: string; rel_path: string }>('SELECT name, rel_path FROM task_artifacts WHERE id = $1 AND task_id = $2', [artifactId, id]);144 if (!art) throw notFound('artifact');145 const res = await rt.raw('GET', '/fs/download', { query: { root: task.cwd_abs, rel: validateRel(art.rel_path, { allowEmpty: false }) } }).catch(runtimeToHttp);146 await audit.req(req, 'task.artifact.download', { target: artifactId });147 return reply148 .header('Content-Type', 'application/octet-stream')149 .header('Content-Disposition', contentDisposition(art.name))150 .header('X-Content-Type-Options', 'nosniff')151 .send(res.body);152 });153}154