Repository Public
1/**2 * Agent tasks. State comes from the runtime (events + reconciler) and is applied through3 * `transition()`, which enforces `canTaskTransition`. Tasks are NEVER retried automatically.4 * Reconciler: runtime unreachable ⇒ nothing changes; runtime healthy and task unknown, or5 * reported LOST ⇒ LOST.6 */7import type { FastifyBaseLogger } from 'fastify';8import { canTaskTransition, type AgentTask, type RtEvent, type RtTaskInfo, type TaskState } from '@cr/shared';9import type { Db, SqlParam } from '../../db.js';10import type { EventHub } from '../../events-hub.js';11import { RuntimeError, type RuntimeClient } from '../../runtime-client.js';12import type { Notifier } from '../notifications/service.js';13import { collectArtifacts } from './artifacts.js';1415export interface TaskRow {16 id: string;17 owner_id: string;18 project_id: string | null;19 title: string;20 command: string;21 host: string;22 cwd_abs: string;23 cwd_rel: string;24 state: TaskState;25 exit_code: number | null;26 timeout_sec: number | null;27 idempotent: boolean;28 retry_of: string | null;29 artifact_globs: string[];30 log_bytes: number;31 cpu_sec: number | null;32 max_rss_mb: number | null;33 created_at: Date;34 started_at: Date | null;35 ended_at: Date | null;36}37export const TERMINAL_TASK_STATES: readonly TaskState[] = ['SUCCEEDED', 'FAILED', 'CANCELLED', 'TIMED_OUT', 'LOST'];38const ACTIVE: readonly TaskState[] = ['RUNNING', 'CANCELLING'];3940export class TaskService {41 constructor(private readonly d: { db: Db; rt: RuntimeClient; hub: EventHub; notifier: Notifier; log: FastifyBaseLogger }) {}4243 owned(id: string, ownerId: string): Promise<TaskRow | null> {44 return this.d.db.one<TaskRow>('SELECT * FROM agent_tasks WHERE id = $1 AND owner_id = $2', [id, ownerId]);45 }46 async dto(r: TaskRow): Promise<AgentTask> {47 const artifacts = await this.d.db.q<{ id: string; name: string; size: number; created_at: Date }>(48 'SELECT id, name, size, created_at FROM task_artifacts WHERE task_id = $1 ORDER BY name', [r.id],49 );50 return {51 id: r.id, projectId: r.project_id, title: r.title, command: r.command, host: r.host, cwd: r.cwd_rel,52 state: r.state, exitCode: r.exit_code, timeoutSec: r.timeout_sec, retryOf: r.retry_of, idempotent: r.idempotent,53 createdAt: r.created_at.toISOString(), startedAt: r.started_at?.toISOString() ?? null,54 endedAt: r.ended_at?.toISOString() ?? null, logBytes: r.log_bytes, cpuSec: r.cpu_sec, maxRssMb: r.max_rss_mb,55 artifacts: artifacts.map((x) => ({ id: x.id, name: x.name, size: x.size, createdAt: x.created_at.toISOString() })),56 };57 }58 async emit(row: TaskRow): Promise<void> {59 this.d.hub.publish(row.owner_id, { type: 'task.updated', task: await this.dto(row) });60 }6162 /** Returns the updated row, or null when the transition is not allowed (nothing written). */63 async transition(id: string, to: TaskState, info?: Partial<RtTaskInfo>): Promise<TaskRow | null> {64 const out = await this.d.db.withTx(async (tx) => {65 const cur = await tx.one<TaskRow>('SELECT * FROM agent_tasks WHERE id = $1 FOR UPDATE', [id]);66 if (!cur) return null;67 if (!canTaskTransition(cur.state, to)) return { refused: cur.state };68 const sets = ['state = $2'];69 const params: SqlParam[] = [id, to];70 const set = (col: string, v: SqlParam): void => {71 params.push(v);72 sets.push(`${col} = $${params.length}`);73 };74 if (info?.exitCode !== undefined) set('exit_code', info.exitCode);75 if (typeof info?.logBytes === 'number') set('log_bytes', info.logBytes);76 if (info?.cpuSec !== undefined) set('cpu_sec', info.cpuSec);77 if (info?.maxRssMb !== undefined) set('max_rss_mb', info.maxRssMb);78 if (to === 'RUNNING' && !cur.started_at) set('started_at', info?.startedAtMs ? new Date(info.startedAtMs) : new Date());79 if (TERMINAL_TASK_STATES.includes(to) && !cur.ended_at) set('ended_at', info?.endedAtMs ? new Date(info.endedAtMs) : new Date());80 const row = await tx.one<TaskRow>(`UPDATE agent_tasks SET ${sets.join(', ')} WHERE id = $1 RETURNING *`, params);81 return { row, from: cur.state };82 });83 if (!out) return null;84 if ('refused' in out) {85 this.d.log.warn({ taskId: id, from: out.refused, to }, 'task transition refused');86 return null;87 }88 if (!out.row) return null;89 const row = out.row;90 if (out.from !== to && TERMINAL_TASK_STATES.includes(to)) await this.onFinished(row);91 await this.emit(row);92 return row;93 }9495 private async onFinished(row: TaskRow): Promise<void> {96 if (row.state !== 'LOST' && row.host === 'local' && row.artifact_globs.length > 0) {97 try {98 await collectArtifacts(this.d.db, this.d.rt, row);99 } catch (err) {100 this.d.log.warn({ err, taskId: row.id }, 'artifact collection failed');101 }102 }103 if (row.state === 'CANCELLED') return;104 const ok = row.state === 'SUCCEEDED';105 await this.d.notifier.notify(row.owner_id, ok ? 'task.succeeded' : 'task.failed', ok ? 'A task succeeded' : 'A task did not complete', row.title, `/tasks/?t=${row.id}`);106 }107108 /** Applies a runtime report. A user-requested cancel keeps CANCELLING until the runtime settles. */109 async applyInfo(info: RtTaskInfo): Promise<void> {110 const cur = await this.d.db.one<{ state: TaskState }>('SELECT state FROM agent_tasks WHERE id = $1', [info.id]);111 if (!cur || TERMINAL_TASK_STATES.includes(cur.state)) return;112 if (info.state === 'RUNNING') {113 if (cur.state === 'RUNNING' || cur.state === 'CANCELLING') {114 await this.d.db.q('UPDATE agent_tasks SET log_bytes = $2, cpu_sec = $3, max_rss_mb = $4 WHERE id = $1', [info.id, info.logBytes, info.cpuSec, info.maxRssMb]);115 return;116 }117 await this.transition(info.id, 'RUNNING', info);118 return;119 }120 // CANCELLING cannot become TIMED_OUT in the state table: record it as CANCELLED.121 const to: TaskState = cur.state === 'CANCELLING' && info.state === 'TIMED_OUT' ? 'CANCELLED' : info.state;122 // The runtime can report the end before our own QUEUED → RUNNING write landed.123 if (cur.state === 'QUEUED' && to !== 'FAILED' && to !== 'CANCELLED') await this.transition(info.id, 'RUNNING', { startedAtMs: info.startedAtMs });124 await this.transition(info.id, to, info);125 }126127 onRuntimeEvent(ev: RtEvent): Promise<void> {128 return ev.type === 'task.updated' ? this.applyInfo(ev.task) : Promise.resolve();129 }130131 /** Returns false (and changes nothing) when the runtime is unreachable. */132 async reconcile(): Promise<boolean> {133 const rows = await this.d.db.q<{ id: string }>('SELECT id FROM agent_tasks WHERE state = ANY($1::text[])', [[...ACTIVE]]);134 if (rows.length === 0) return true;135 let infos: RtTaskInfo[];136 try {137 infos = await this.d.rt.json<RtTaskInfo[]>('GET', '/tasks', undefined, { timeoutMs: 5000 });138 if (!Array.isArray(infos)) return false;139 } catch (err) {140 if (!(err instanceof RuntimeError)) throw err;141 return false;142 }143 const byId = new Map(infos.map((i) => [i.id, i]));144 for (const { id } of rows) {145 try {146 const info = byId.get(id);147 if (info) await this.applyInfo(info);148 else await this.transition(id, 'LOST');149 } catch (err) {150 this.d.log.error({ err, taskId: id }, 'task reconcile failed');151 }152 }153 return true;154 }155}156