/** * agent/runner.mjs — Agent 작업 실행기(단계 저장·재개·진행 보고). * * HAL 의 lib/jobs 와 같은 원칙: 작업은 재진입 가능하고 끝난 단계는 건너뛴다. 단계 상태는 data/jobs/.json 에 저장되어 Agent 재시작 뒤 이어서 실행된다. * ctx = { job, payload, roots, sys, client, dataDir, step(id, fn, title), log(line), progress(pct), needsUser(message, hint), checkCancel(), steps() } * 결과 보고: 성공 { status:'done', result } / 실패 { status:'failed', error, errorCode } / 사용자·운영자 확인 { status:'needs_user', userMessage }. */ import { readFileSync, writeFileSync, mkdirSync, existsSync, unlinkSync, renameSync } from 'node:fs'; import { join } from 'node:path'; import { redact, safeError } from '../lib/redact.mjs'; import { NeedsUser } from './sys.mjs'; class Signal extends Error { constructor(type, payload) { super(type); this.signal = type; this.payload = payload || {}; } } export function createRunner({ kinds, roots, sys, client, dataDir, log = () => {} }) { const jobsDir = join(dataDir, 'jobs'); mkdirSync(jobsDir, { recursive: true }); const stateFile = (id) => join(jobsDir, `${id}.json`); const loadState = (id) => { try { return JSON.parse(readFileSync(stateFile(id), 'utf8')); } catch (_) { return { steps: [], log: [] }; } }; const saveState = (id, st) => { const tmp = stateFile(id) + '.tmp'; writeFileSync(tmp, JSON.stringify(st)); renameSync(tmp, stateFile(id)); }; /** 진행 보고(HAL). 실패해도 작업은 계속한다(다음 보고에서 따라잡음). */ async function report(id, body) { if (!client) return null; try { return await client.request('POST', `/agent/v1/jobs/${id}/progress`, body); } catch (e) { log(`[agent] 진행 보고 실패 ${id}: ${safeError(e)}`); return null; } } async function execute(job) { const def = kinds[job.kind]; const st = loadState(job.id); let cancelled = false; const ctx = { job, payload: job.payload || {}, roots, sys, client, dataDir, kinds, async step(id, fn, title) { let s = st.steps.find((x) => x.id === id); if (!s) { s = { id, title: title || id, state: 'pending' }; st.steps.push(s); } if (s.state === 'done') return s.result; s.state = 'running'; s.startedAt = new Date().toISOString(); delete s.error; saveState(job.id, st); await report(job.id, { steps: st.steps }); try { const r = await fn(); s.state = 'done'; s.endedAt = new Date().toISOString(); s.result = r === undefined ? null : redact(r); saveState(job.id, st); return r; } catch (e) { s.state = e instanceof Signal ? 'needs_user' : 'failed'; s.error = safeError(e); s.endedAt = new Date().toISOString(); saveState(job.id, st); throw e; } }, log(line) { st.log.push({ at: new Date().toISOString(), line: String(line).slice(0, 500) }); st.log = st.log.slice(-200); saveState(job.id, st); log(`[${job.kind} ${job.id}] ${line}`); }, async progress(pct) { st.progress = Math.max(0, Math.min(100, Math.round(pct))); saveState(job.id, st); const r = await report(job.id, { progress: st.progress, steps: st.steps, log: st.log.slice(-20) }); if (r && r.cancelRequested) cancelled = true; }, needsUser(message, hint) { throw new Signal('needs_user', { message, hint }); }, checkCancel() { if (cancelled) throw new Signal('cancel', {}); }, steps: () => st.steps, state: st, }; let out; if (!def) out = { status: 'failed', error: `알 수 없는 작업 종류: ${job.kind}`, errorCode: 'unknown_kind' }; else { try { const result = await def.run(ctx); out = { status: 'done', result: redact(result === undefined ? {} : result), steps: st.steps, log: st.log.slice(-50) }; } catch (e) { if (e instanceof Signal && e.signal === 'needs_user') out = { status: 'needs_user', userMessage: `${e.payload.message}${e.payload.hint ? ` (${e.payload.hint})` : ''}`, steps: st.steps, log: st.log.slice(-50) }; else if (e instanceof Signal && e.signal === 'cancel') out = { status: 'failed', error: '취소 요청으로 중단', errorCode: 'cancelled', steps: st.steps, log: st.log.slice(-50) }; else if (e instanceof NeedsUser || e.needsUser) out = { status: 'needs_user', userMessage: `${e.message}${e.hint ? ` — ${e.hint}` : ''}`, steps: st.steps, log: st.log.slice(-50) }; else out = { status: 'failed', error: safeError(e), errorCode: e.code || 'failed', steps: st.steps, log: st.log.slice(-50), result: e.partial ? redact(e.partial) : undefined }; } } try { unlinkSync(stateFile(job.id)); } catch (_) {} if (client) { try { await client.request('POST', `/agent/v1/jobs/${job.id}/result`, out); } catch (e) { log(`[agent] 결과 보고 실패 ${job.id}: ${safeError(e)}`); saveState(job.id, { ...st, pendingResult: out }); } } return out; } /** 재시작 뒤: 보고하지 못한 결과가 남아 있으면 다시 보고 */ async function flushPending() { if (!client) return; const { readdirSync } = await import('node:fs'); for (const f of readdirSync(jobsDir).filter((x) => x.endsWith('.json'))) { const st = loadState(f.slice(0, -5)); if (st.pendingResult) { try { await client.request('POST', `/agent/v1/jobs/${f.slice(0, -5)}/result`, st.pendingResult); unlinkSync(join(jobsDir, f)); } catch (_) {} } } } return { execute, loadState, flushPending, kinds }; }