Release v2.2.1: Autonomous Task Resumption & Engine Resilience

2026-05-14 23:27:51 +09:00
parent e86e3177c7
commit cd22da8735
21 changed files with 848 additions and 126 deletions
@@ -57,8 +57,14 @@ import {
    writeReport,
    writeSessionJson,
 } from './sessionStore';
+import {
+    markResumeStatus,
+    readResumeState,
+    resolveSessionDir,
+    writeResumeState,
+} from './resumeStore';
 import { buildTelegramReporter, formatCompanyTelegramReport } from './telegramReport';
-import { AgentTurnOutput, CompanyState, CompanyTaskPlan, PipelineDef, PipelineStage, SessionResult } from './types';
+import { AgentTurnOutput, CompanyResumeState, CompanyState, CompanyTaskPlan, PipelineDef, PipelineStage, SessionResult } from './types';

 /** Trim length applied when an agent's output is fed into the next agent. */
 const PEER_OUTPUT_BUDGET = 1500;
@@ -159,72 +165,190 @@ export interface DispatcherDeps {
 /**
 * Run a single company turn. Returns a fully-populated `SessionResult` even
 * on partial failure (so callers can always render *something* in chat).
+ *
+ * When `seed` is supplied, this is a *resume* of a previously-aborted turn:
+ * the saved sessionDir/plan/outputs are reused, and dispatch picks up at the
+ * saved `nextIndex` instead of starting from scratch. The CEO planner is
+ * skipped on resume — we trust the original plan to keep behaviour
+ * deterministic. Pipeline mode also restores `latestByStage` / `iterations`
+ * / `revisionNotes` so loop-backs and template tokens resolve as if the
+ * turn never paused.
 */
 export async function runCompanyTurn(
    userPrompt: string,
    deps: DispatcherDeps,
+    seed?: CompanyResumeState | null,
 ): Promise<SessionResult> {
    const startedAt = Date.now();
    const state = readCompanyState(deps.context);
-    const timestamp = newSessionTimestamp();
-    const sessionDir = createSessionDir(deps.context, timestamp);
+    const timestamp = seed?.timestamp ?? newSessionTimestamp();
+    const sessionDir = seed
+        ? resolveSessionDir(deps.context, seed.timestamp)
+        : createSessionDir(deps.context, timestamp);
+    const startedAtIso = seed?.startedAt ?? new Date().toISOString();

    const emit: CompanyTurnEmitter = deps.onEvent ?? (() => { /* noop */ });
    const isAborted = () => deps.signal?.aborted === true;
-    const fail = (reason: string): SessionResult => {
+
+    // ── Resume state writer ──
+    // 모든 의미 있는 시점(plan 확정 / 각 stage 직후 / abort)에 같은 파일을 덮어쓴다.
+    // dispatch 중에 캡처해야 할 cursor·캐시들은 클로저로 넘기는 게 매번
+    // 인자를 6개씩 줄지어 보내는 것보다 깔끔하다.
+    const persistResume = (
+        status: CompanyResumeState['status'],
+        partial: {
+            plan: CompanyTaskPlan;
+            pipelineId: string | null;
+            outputs: AgentTurnOutput[];
+            nextIndex: number;
+            pipelineContext?: CompanyResumeState['pipelineContext'];
+            abortReason?: string;
+        },
+    ): void => {
+        writeResumeState(sessionDir, {
+            version: 1,
+            timestamp,
+            userPrompt,
+            pipelineId: partial.pipelineId,
+            plan: partial.plan,
+            agentOutputs: partial.outputs,
+            nextIndex: partial.nextIndex,
+            pipelineContext: partial.pipelineContext,
+            status,
+            abortReason: partial.abortReason,
+            lastUpdatedAt: new Date().toISOString(),
+            startedAt: startedAtIso,
+        });
+    };
+
+    const fail = (reason: string, ctx?: {
+        plan: CompanyTaskPlan;
+        pipelineId: string | null;
+        outputs: AgentTurnOutput[];
+        nextIndex: number;
+        pipelineContext?: CompanyResumeState['pipelineContext'];
+    }): SessionResult => {
        emit({ phase: 'aborted', reason });
+        // abort 시점의 상태를 _resume.json에 영구화 — 이후 사용자가 "이어서 진행"
+        // 누르면 이 파일에서 plan + 진행도를 복원해 nextIndex부터 재개 가능.
+        if (ctx) {
+            persistResume('aborted', { ...ctx, abortReason: reason });
+        } else if (seed) {
+            // Resume 도중에 즉시 abort — 들어왔던 seed를 그대로 abort로 다시 마킹.
+            markResumeStatus(sessionDir, 'aborted', reason);
+        }
        return {
            timestamp, sessionDir,
            userPrompt,
-            plan: { brief: '', tasks: [] },
-            agentOutputs: [],
+            plan: ctx?.plan ?? seed?.plan ?? { brief: '', tasks: [] },
+            agentOutputs: ctx?.outputs ?? seed?.agentOutputs ?? [],
            report: '',
            totalDurationMs: Date.now() - startedAt,
        };
    };
    if (isAborted()) return fail('signal-aborted');

-    // ── Phase 1: plan (pipeline or legacy planner) ──
-    emit({ phase: 'plan-start' });
-    const pipeline = resolveActivePipeline(state);
+    // ── Phase 1: plan (pipeline or legacy planner — skipped on resume) ──
+    let pipeline: PipelineDef | null;
    let plan: CompanyTaskPlan;
    let plannerRaw = '';
-    let plannerParsed = false;
-    if (pipeline) {
-        // Pipeline mode: the user has authored a fixed sequence of stages.
-        // We still surface a `plan` for the report writer and the session
-        // summary — derived directly from the pipeline definition.
-        plan = {
-            brief: `[Pipeline: ${pipeline.name}] ${userPrompt.slice(0, 200)}`,
-            tasks: pipeline.stages.map((s) => ({ agent: s.agentId, task: s.label })),
-        };
-        plannerParsed = true;
+    let plannerParsed = true;
+    if (seed) {
+        // Resume path: reuse the original plan + pipeline binding verbatim.
+        // If the user edited the pipeline definition in the meantime we still
+        // re-resolve by id from the current state — that way deleted stages
+        // wouldn't crash the dispatch (resolveActivePipeline returns null
+        // gracefully). Worst case the dispatch finishes the leftover plan.
+        pipeline = seed.pipelineId
+            ? (state.pipelines?.[seed.pipelineId] ?? null)
+            : null;
+        plan = seed.plan;
+        emit({ phase: 'plan-start' });
+        emit({ phase: 'plan-ready', plan, parsed: true, raw: '' });
    } else {
-        const ceoModel = modelForAgent(state, 'ceo', deps.defaultModel);
-        const plannerResult = await runCeoPlanner(deps.ai, userPrompt, state, { model: ceoModel });
-        plan = plannerResult.plan;
-        plannerRaw = plannerResult.raw;
-        plannerParsed = plannerResult.parsed;
+        emit({ phase: 'plan-start' });
+        pipeline = resolveActivePipeline(state);
+        if (pipeline) {
+            // Pipeline mode: the user has authored a fixed sequence of stages.
+            // We still surface a `plan` for the report writer and the session
+            // summary — derived directly from the pipeline definition.
+            plan = {
+                brief: `[Pipeline: ${pipeline.name}] ${userPrompt.slice(0, 200)}`,
+                tasks: pipeline.stages.map((s) => ({ agent: s.agentId, task: s.label })),
+            };
+        } else {
+            const ceoModel = modelForAgent(state, 'ceo', deps.defaultModel);
+            const plannerResult = await runCeoPlanner(deps.ai, userPrompt, state, { model: ceoModel });
+            plan = plannerResult.plan;
+            plannerRaw = plannerResult.raw;
+            plannerParsed = plannerResult.parsed;
+        }
+        if (isAborted()) {
+            return fail('aborted-after-plan', {
+                plan, pipelineId: pipeline?.id ?? null, outputs: [], nextIndex: 0,
+            });
+        }
+        emit({ phase: 'plan-ready', plan, parsed: plannerParsed, raw: plannerRaw });
+        writeBrief(sessionDir, userPrompt, plan);
    }
-    if (isAborted()) return fail('aborted-after-plan');
-    emit({
-        phase: 'plan-ready',
-        plan,
-        parsed: plannerParsed,
-        raw: plannerRaw,
+    const pipelineId = pipeline?.id ?? null;
+
+    // 초기 resume 상태(또는 resume 진행 시작 상태) 영속화.
+    persistResume('in-progress', {
+        plan, pipelineId,
+        outputs: seed?.agentOutputs ?? [],
+        nextIndex: seed?.nextIndex ?? 0,
+        pipelineContext: seed?.pipelineContext,
    });
-    writeBrief(sessionDir, userPrompt, plan);

    // ── Phase 2: sequential dispatch ──
-    const outputs: AgentTurnOutput[] = [];
+    const outputs: AgentTurnOutput[] = seed?.agentOutputs ? [...seed.agentOutputs] : [];
    if (pipeline) {
-        const runResult = await _runPipeline(pipeline, userPrompt, plan.brief, sessionDir, timestamp, state, deps, isAborted, emit);
-        if (runResult.aborted) return fail(runResult.aborted);
-        outputs.push(...runResult.outputs);
+        const runResult = await _runPipeline(
+            pipeline, userPrompt, plan.brief, sessionDir, timestamp, state, deps, isAborted, emit,
+            seed && seed.pipelineId === pipeline.id
+                ? {
+                    outputs,
+                    latestByStage: seed.pipelineContext?.latestByStage ?? {},
+                    iterations: seed.pipelineContext?.iterations ?? {},
+                    revisionNotes: seed.pipelineContext?.revisionNotes ?? {},
+                    startIndex: seed.nextIndex,
+                }
+                : undefined,
+            (commit) => {
+                // 각 stage 직후 호출됨 — _runPipeline의 내부 cursor·캐시를 통째로 받아
+                // resume 파일을 갱신. 다음 stage가 시작되기 전에 디스크에 한 번 떨어짐.
+                persistResume('in-progress', {
+                    plan, pipelineId,
+                    outputs: commit.outputs,
+                    nextIndex: commit.nextIndex,
+                    pipelineContext: {
+                        latestByStage: commit.latestByStage,
+                        iterations: commit.iterations,
+                        revisionNotes: commit.revisionNotes,
+                    },
+                });
+            },
+        );
+        if (runResult.aborted) {
+            return fail(runResult.aborted, {
+                plan, pipelineId,
+                outputs: runResult.outputs,
+                nextIndex: runResult.nextIndex ?? outputs.length,
+                pipelineContext: runResult.pipelineContext,
+            });
+        }
+        // outputs는 이미 _runPipeline 내부에서 누적 — 새 객체 push 불필요.
+        outputs.length = 0; outputs.push(...runResult.outputs);
    } else {
        const total = plan.tasks.length;
-        for (let i = 0; i < total; i++) {
-            if (isAborted()) return fail('aborted-mid-dispatch');
+        const startIdx = seed?.nextIndex ?? 0;
+        for (let i = startIdx; i < total; i++) {
+            if (isAborted()) {
+                return fail('aborted-mid-dispatch', {
+                    plan, pipelineId: null, outputs, nextIndex: i,
+                });
+            }
            const task = plan.tasks[i];
            emit({ phase: 'agent-start', agentId: task.agent, task: task.task, index: i, total });
            const turn = await _dispatchOne(task.agent, task.task, outputs, state, deps);
@@ -236,11 +360,25 @@ export async function runCompanyTurn(
                `[${timestamp}] ${task.task} — ${turn.error ? `❌ ${turn.error}` : '✅'}`,
            );
            emit({ phase: 'agent-done', agentId: task.agent, output: turn, index: i, total });
+            // 각 task 직후 resume cursor 영속화 — 다음 task 전에 abort/crash 나도
+            // 같은 task 중복 실행 없이 그 다음부터 이어진다.
+            persistResume('in-progress', {
+                plan, pipelineId: null, outputs, nextIndex: i + 1,
+            });
        }
    }

    // ── Phase 3: synthesis ──
-    if (isAborted()) return fail('aborted-before-report');
+    // 여기까지 왔다는 건 모든 stage/task가 끝난 상태. abort 시에도 outputs는
+    // 이미 disk에 떨어져 있으므로 resume cursor를 plan.tasks.length로 옮겨
+    // 다음 재개 시점에는 report 단계부터 시작하도록 한다 (= 사실상 완료에 가깝다).
+    if (isAborted()) {
+        return fail('aborted-before-report', {
+            plan, pipelineId,
+            outputs,
+            nextIndex: plan.tasks.length,
+        });
+    }
    emit({ phase: 'report-start' });
    const reportModel = modelForAgent(state, 'ceo', deps.defaultModel);
    const reportResult = await runCeoReporter(
@@ -293,6 +431,9 @@ export async function runCompanyTurn(
        totalDurationMs: Date.now() - startedAt,
    };
    writeSessionJson(sessionDir, result);
+    // 자연 종료 — resume 파일은 'completed' 마킹으로 listResumable에서 자동 제외.
+    // 파일은 삭제하지 않고 남겨 추후 감사/디버깅 용도로 활용 (수 KB 수준).
+    markResumeStatus(sessionDir, 'completed');
    // Heuristic: if the report mentions a 🚀 line, extract it as a decision.
    const decisionLine = reportResult.report.split(/\n/).find((l) => /^\d+\.\s+/.test(l.trim()));
    if (decisionLine) appendDecision(deps.context, decisionLine.trim());
@@ -305,6 +446,43 @@ export async function runCompanyTurn(
    return result;
 }

+/**
+ * Resume a previously-aborted company turn from disk.
+ *
+ * sessionDir의 `_resume.json`을 읽어 plan + 진행 cursor + 캐시를 복원한 다음
+ * `runCompanyTurn`을 seed 옵션으로 호출. 같은 sessionDir을 재사용하므로
+ * markdown 산출물(`_brief.md`, `<agentId>.md` 등)이 누적됨.
+ *
+ * @returns 복구 가능한 상태가 있으면 SessionResult, 아니면 null
+ *   (`_resume.json`이 없거나, 이미 'completed' 상태이거나, plan이 비어 있는 경우).
+ */
+export async function resumeCompanyTurn(
+    timestamp: string,
+    deps: DispatcherDeps,
+): Promise<SessionResult | null> {
+    const sessionDir = resolveSessionDir(deps.context, timestamp);
+    const saved = readResumeState(sessionDir);
+    if (!saved) {
+        logInfo('company.dispatcher: resume requested but no state found.', { timestamp });
+        return null;
+    }
+    if (saved.status === 'completed') {
+        logInfo('company.dispatcher: resume requested for completed session — ignoring.', { timestamp });
+        return null;
+    }
+    if (!saved.plan || !Array.isArray(saved.plan.tasks) || saved.plan.tasks.length === 0) {
+        logInfo('company.dispatcher: resume requested but plan is empty.', { timestamp });
+        return null;
+    }
+    logInfo('company.dispatcher: resuming turn.', {
+        timestamp,
+        pipelineId: saved.pipelineId,
+        nextIndex: saved.nextIndex,
+        totalTasks: saved.plan.tasks.length,
+    });
+    return runCompanyTurn(saved.userPrompt, deps, saved);
+}
+
 /**
 * Dispatch one specialist. Wraps the AI call with try/catch so a single
 * agent's failure never aborts the whole turn — we record the error and
@@ -472,6 +650,28 @@ async function _dispatchOne(
 * Returns `{ outputs, aborted }`: `aborted` is set only when the abort
 * signal flipped mid-run; the outer dispatcher then short-circuits.
 */
+interface PipelineSeed {
+    /** 이미 끝낸 stage들의 출력. 새로운 outputs 배열의 시드로 들어감. */
+    outputs: AgentTurnOutput[];
+    /** stage id → 최신 출력. `{{stage.<id>}}` 치환에 사용. */
+    latestByStage: Record<string, AgentTurnOutput>;
+    /** stage id → loop-back 누적 횟수. */
+    iterations: Record<string, number>;
+    /** stage id → 사용자 수정요청 코멘트. */
+    revisionNotes: Record<string, string>;
+    /** 재개를 시작할 0-based stage index. */
+    startIndex: number;
+}
+
+/** _runPipeline이 매 stage 직후 호출하는 commit 콜백의 payload. */
+export interface PipelineCommit {
+    outputs: AgentTurnOutput[];
+    latestByStage: Record<string, AgentTurnOutput>;
+    iterations: Record<string, number>;
+    revisionNotes: Record<string, string>;
+    nextIndex: number;
+}
+
 async function _runPipeline(
    pipeline: PipelineDef,
    userPrompt: string,
@@ -482,21 +682,58 @@ async function _runPipeline(
    deps: DispatcherDeps,
    isAborted: () => boolean,
    emit: CompanyTurnEmitter,
-): Promise<{ outputs: AgentTurnOutput[]; aborted?: string }> {
-    const outputs: AgentTurnOutput[] = [];
+    seed?: PipelineSeed,
+    onStageCommit?: (commit: PipelineCommit) => void,
+): Promise<{
+    outputs: AgentTurnOutput[];
+    aborted?: string;
+    /** abort 시점의 stage cursor — runCompanyTurn이 resume 파일 영속화에 사용. */
+    nextIndex?: number;
+    pipelineContext?: {
+        latestByStage: Record<string, AgentTurnOutput>;
+        iterations: Record<string, number>;
+        revisionNotes: Record<string, string>;
+    };
+}> {
+    const outputs: AgentTurnOutput[] = seed?.outputs ? [...seed.outputs] : [];
    // Keep the latest output per stage id so `{{stage.<id>}}` template
    // tokens always resolve to the most recent value across loop-backs.
-    const latestByStage: Record<string, AgentTurnOutput> = {};
-    const iterations: Record<string, number> = {};
+    const latestByStage: Record<string, AgentTurnOutput> = seed?.latestByStage
+        ? { ...seed.latestByStage }
+        : {};
+    const iterations: Record<string, number> = seed?.iterations
+        ? { ...seed.iterations }
+        : {};
    const total = pipeline.stages.length;
    // Per-stage extra instruction injected by user revision requests. Cleared
    // after the stage re-runs successfully so it doesn't pollute the rest of
    // the pipeline.
-    const revisionNotes: Record<string, string> = {};
-    let i = 0;
-    let stepIndex = 0;
+    const revisionNotes: Record<string, string> = seed?.revisionNotes
+        ? { ...seed.revisionNotes }
+        : {};
+    let i = seed?.startIndex ?? 0;
+    // stepIndex는 emit의 index 인자(UI 진행률) — 재개 시 이미 완료된 stage 수만큼
+    // 미리 진행시켜둬야 "참여 중인 stage 4/7" 같은 표시가 정확해진다.
+    let stepIndex = seed?.outputs?.length ?? 0;
+    /** Snapshot helper — 현재 cursor·캐시 묶음을 PipelineCommit으로 캡슐. */
+    const snapshot = (nextIdx: number): PipelineCommit => ({
+        outputs: [...outputs],
+        latestByStage: { ...latestByStage },
+        iterations: { ...iterations },
+        revisionNotes: { ...revisionNotes },
+        nextIndex: nextIdx,
+    });
+    const abortReturn = (reason: string) => ({
+        outputs, aborted: reason,
+        nextIndex: i,
+        pipelineContext: {
+            latestByStage: { ...latestByStage },
+            iterations: { ...iterations },
+            revisionNotes: { ...revisionNotes },
+        },
+    });
    while (i < pipeline.stages.length) {
-        if (isAborted()) return { outputs, aborted: 'aborted-mid-pipeline' };
+        if (isAborted()) return abortReturn('aborted-mid-pipeline');
        const stage = pipeline.stages[i];
        const baseTask = _renderStageInstruction(stage, userPrompt, brief, latestByStage);
        const note = revisionNotes[stage.id];
@@ -539,14 +776,16 @@ async function _runPipeline(
                // 호스트가 에러를 던지면 안전하게 중단 — 무한 대기 방지.
                decision = { kind: 'abort' };
            }
-            if (isAborted()) return { outputs, aborted: 'aborted-mid-approval' };
+            if (isAborted()) return abortReturn('aborted-mid-approval');
            emit({ phase: 'approval-resolved', stageId: stage.id, decision: decision.kind });
            if (decision.kind === 'abort') {
-                return { outputs, aborted: 'aborted-by-user-at-approval' };
+                return abortReturn('aborted-by-user-at-approval');
            }
            if (decision.kind === 'revise') {
                revisionNotes[stage.id] = decision.comment || '(추가 코멘트 없음)';
                // 같은 stage 재실행 — i를 그대로 두고 continue.
+                // 수정요청 코멘트도 resume에 반영해야 재개 시 사용자 의도 보존.
+                onStageCommit?.(snapshot(i));
                continue;
            }
            // 'approve' → 아래 loop-back/다음 stage 진행 로직으로 자연히 fall-through.
@@ -565,11 +804,15 @@ async function _runPipeline(
                if (targetIdx !== -1 && targetIdx < i) {
                    emit({ phase: 'stage-loop', from: stage.id, to: stage.loopBackTo, iteration: count });
                    i = targetIdx;
+                    onStageCommit?.(snapshot(i));
                    continue;
                }
            }
        }
        i++;
+        // 매 stage 자연 완료 직후 resume 파일을 갱신 — 다음 stage 시작 전에 디스크에
+        // 떨어지므로 그 사이에 abort/crash가 나도 정확히 그 다음 stage부터 재개된다.
+        onStageCommit?.(snapshot(i));
    }
    return { outputs };
 }