maestro/test/ingest.test.ts

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { Store } from '../src/store/index.js';
import { ingestRun } from '../src/daemon/ingest.js';
import { appendOutbox, type OutboxPayload } from '../src/executor/protocol.js';
import { branchFor, worktreeDirFor } from '../src/executor/worktree.js';

const noopLog = { info: (): void => undefined, error: (): void => undefined };

/**
 * 隔离的 MAESTRO_DATA_DIR（outbox 落在临时目录），跑回调，结束清理。
 * protocol.ts 的 runDir/outboxPath 在调用时读 env，故进 cb 前设好、cb 内的 append/ingest 都命中临时目录。
 */
function withTmpDataDir(cb: () => void): void {
  const dir = mkdtempSync(join(tmpdir(), 'maestro-ingest-'));
  const prev = process.env.MAESTRO_DATA_DIR;
  process.env.MAESTRO_DATA_DIR = dir;
  try {
    cb();
  } finally {
    if (prev === undefined) delete process.env.MAESTRO_DATA_DIR;
    else process.env.MAESTRO_DATA_DIR = prev;
    rmSync(dir, { recursive: true, force: true });
  }
}

/** 真 Store + 项目 + easy 任务推进到 executing，并建一条 executor run（模拟 daemon 已领取、worker 已 spawn）。 */
function setupExecuting(opts: { autonomy?: 'auto-easy' | 'auto-approved'; maxRetries?: number } = {}): {
  store: Store; projectId: string; taskId: string; runId: string; dir: string; branch: string;
} {
  const store = new Store(':memory:');
  const p = store.createProject({
    name: 'ingest', repoPath: '/tmp/ingest-repo-' + Math.random(),
    autonomy: opts.autonomy ?? 'auto-easy', maxRetries: opts.maxRetries,
  });
  const t = store.createTask({ projectId: p.id, title: 'tweak', complexity: 'easy' });
  store.setOperations(t.id, '在 README.md 追加一行');
  const branch = branchFor(t.id);
  const dir = worktreeDirFor(p.repoPath, t.id);
  store.transition(t.id, 'queued', { by: 'test' });
  store.transition(t.id, 'executing', { by: 'test' });
  const run = store.startRun(t.id, 'executor', { worktree: dir, branch });
  store.setWorkerPid(run.id, 4242);
  return { store, projectId: p.id, taskId: t.id, runId: run.id, dir, branch };
}

function resultPayload(branch: string, worktree: string): OutboxPayload {
  return {
    type: 'result',
    branch, worktree,
    diffSummary: ' README.md | 1 +',
    commits: ['abc1234 hello maestro'],
    executor: { transcriptRef: '/tmp/exec.jsonl', sessionId: 'sess-exec-1' },
    code: { summary: '## 做了什么\nmock 复审通过', verdict: 'approve', transcriptRef: '/tmp/code.jsonl' },
    security: { summary: '## 安全审计\nmock 审计通过', verdict: 'approve', transcriptRef: '/tmp/sec.jsonl' },
  };
}

test('ingest result：task→exec_review，setResult 四字段正确，reviewer/security/executor 各 1 条 succeeded', () => {
  withTmpDataDir(() => {
    const { store, taskId, runId, dir, branch } = setupExecuting();

    appendOutbox(runId, { type: 'started', pid: 4242, worktree: dir, branch, model: 'claude-sonnet-4-6' });
    appendOutbox(runId, { type: 'phase', phase: 'executing' });
    appendOutbox(runId, resultPayload(branch, dir));
    appendOutbox(runId, { type: 'done' });

    ingestRun(store, noopLog, runId);

    const done = store.getTask(taskId)!;
    assert.equal(done.status, 'exec_review');
    assert.deepEqual(done.result, {
      branch,
      worktree: dir,
      diffSummary: ' README.md | 1 +',
      commits: ['abc1234 hello maestro'],
      prUrl: null,
      summary: '## 做了什么\nmock 复审通过',
      verdict: 'approve',
      securitySummary: '## 安全审计\nmock 审计通过',
      securityVerdict: 'approve',
      mergeTaskId: null,
    });

    const runs = store.listRuns(taskId);
    const executor = runs.find((r) => r.kind === 'executor')!;
    assert.equal(executor.status, 'succeeded');
    assert.equal(executor.transcriptRef, '/tmp/exec.jsonl');
    assert.equal(executor.claudeSessionId, 'sess-exec-1');
    const reviewer = runs.filter((r) => r.kind === 'reviewer');
    assert.equal(reviewer.length, 1);
    assert.equal(reviewer[0].status, 'succeeded');
    assert.equal(reviewer[0].transcriptRef, '/tmp/code.jsonl');
    const security = runs.filter((r) => r.kind === 'security');
    assert.equal(security.length, 1);
    assert.equal(security[0].status, 'succeeded');
    assert.equal(security[0].transcriptRef, '/tmp/sec.jsonl');

    // 游标推进到末条（done.seq=4）
    assert.equal(store.getRun(runId)!.lastSeq, 4);
    store.close();
  });
});

test('ingest result：任一 verdict=reject 也照常落进 result（裁决归用户）', () => {
  withTmpDataDir(() => {
    const { store, taskId, runId, dir, branch } = setupExecuting();
    appendOutbox(runId, {
      ...resultPayload(branch, dir),
      code: { summary: '发现问题', verdict: 'reject', transcriptRef: '/tmp/code.jsonl' },
      security: { summary: '发现密钥泄露', verdict: 'reject', transcriptRef: '/tmp/sec.jsonl' },
    } as OutboxPayload);
    ingestRun(store, noopLog, runId);

    const done = store.getTask(taskId)!;
    assert.equal(done.status, 'exec_review');
    assert.equal(done.result!.verdict, 'reject');
    assert.equal(done.result!.summary, '发现问题');
    assert.equal(done.result!.securityVerdict, 'reject');
    assert.equal(done.result!.securitySummary, '发现密钥泄露');
    store.close();
  });
});

test('ingest failed：首次失败 → task 转 queued + 持久化退避（nextEligibleAt 有值）+ executor run failed', () => {
  withTmpDataDir(() => {
    const { store, taskId, runId } = setupExecuting();

    appendOutbox(runId, { type: 'started', pid: 4242, worktree: '/x', branch: 'b', model: 'm' });
    appendOutbox(runId, { type: 'failed', error: 'boom #1', transcriptRef: '/tmp/f.jsonl', sessionId: 'sess-f-1' });
    appendOutbox(runId, { type: 'done' });

    ingestRun(store, noopLog, runId);

    const t = store.getTask(taskId)!;
    assert.equal(t.status, 'queued');                       // 第一次失败 → 重入队（默认 maxRetries=2）
    assert.ok(t.nextEligibleAt, 'nextEligibleAt 应被持久化（退避）');
    assert.ok(Date.parse(t.nextEligibleAt!) > Date.now() - 1000, 'nextEligibleAt 在未来');

    const executor = store.listRuns(taskId).find((r) => r.kind === 'executor')!;
    assert.equal(executor.status, 'failed');
    assert.match(executor.error ?? '', /boom #1/);
    assert.equal(executor.transcriptRef, '/tmp/f.jsonl');
    assert.equal(executor.claudeSessionId, 'sess-f-1');
    store.close();
  });
});

test('ingest failed：maxRetries=0 → 首次失败直接 needs_attention', () => {
  withTmpDataDir(() => {
    const { store, taskId, runId } = setupExecuting({ maxRetries: 0 });
    appendOutbox(runId, { type: 'failed', error: 'boom', transcriptRef: null, sessionId: null });
    ingestRun(store, noopLog, runId);
    assert.equal(store.getTask(taskId)!.status, 'needs_attention');
    store.close();
  });
});

test('幂等：同一批 outbox ingest 两次，DB 不重复变更（lastSeq 生效）', () => {
  withTmpDataDir(() => {
    const { store, taskId, runId, dir, branch } = setupExecuting();
    appendOutbox(runId, { type: 'started', pid: 4242, worktree: dir, branch, model: 'm' });
    appendOutbox(runId, resultPayload(branch, dir));
    appendOutbox(runId, { type: 'done' });

    ingestRun(store, noopLog, runId);
    const after1 = store.getTask(taskId)!;
    const runs1 = store.listRuns(taskId);
    assert.equal(after1.status, 'exec_review');
    assert.equal(runs1.filter((r) => r.kind === 'reviewer').length, 1);
    assert.equal(runs1.filter((r) => r.kind === 'security').length, 1);
    const lastSeq1 = store.getRun(runId)!.lastSeq;

    // 第二次 ingest：seq 全部 ≤ lastSeq → 不处理任何记录
    ingestRun(store, noopLog, runId);
    const runs2 = store.listRuns(taskId);
    assert.equal(runs2.length, runs1.length, '复审 run 不应重复创建');
    assert.equal(runs2.filter((r) => r.kind === 'reviewer').length, 1);
    assert.equal(runs2.filter((r) => r.kind === 'security').length, 1);
    assert.equal(store.getRun(runId)!.lastSeq, lastSeq1, 'lastSeq 不变');
    assert.equal(store.getTask(taskId)!.status, 'exec_review');
    store.close();
  });
});

test('ingest 续读：先 ingest 半截（started/phase），再追加 result，第二次 ingest 完成 exec_review', () => {
  withTmpDataDir(() => {
    const { store, taskId, runId, dir, branch } = setupExecuting();
    appendOutbox(runId, { type: 'started', pid: 4242, worktree: dir, branch, model: 'm' });
    appendOutbox(runId, { type: 'phase', phase: 'executing' });
    ingestRun(store, noopLog, runId);
    assert.equal(store.getTask(taskId)!.status, 'executing');   // 还没到 result，仍 executing
    assert.equal(store.getRun(runId)!.lastSeq, 2);

    appendOutbox(runId, resultPayload(branch, dir));
    appendOutbox(runId, { type: 'done' });
    ingestRun(store, noopLog, runId);
    assert.equal(store.getTask(taskId)!.status, 'exec_review');
    assert.equal(store.getRun(runId)!.lastSeq, 4);
    store.close();
  });
});

// ───────────────────────── planner run 摄取 ─────────────────────────

/** 真 Store + 一个 medium(speccing) 或 hard(analyzing) 任务 + 一条 planner run（模拟已领取规划）。 */
function setupPlanning(complexity: 'medium' | 'hard'): { store: Store; taskId: string; runId: string } {
  const store = new Store(':memory:');
  const p = store.createProject({ name: 'plan', repoPath: '/tmp/plan-' + Math.random(), autonomy: 'auto-approved' });
  const t = store.createTask({ projectId: p.id, title: '大任务', complexity });
  // 建任务即落 speccing(medium)/analyzing(hard)，不转状态，直接起 planner run
  const run = store.startRun(t.id, 'planner', { worktree: p.repoPath, branch: 'n/a' });
  store.setWorkerPid(run.id, 4243);
  return { store, taskId: t.id, runId: run.id };
}

test('ingest planner-spec：setSpec + 转 spec_review + 收尾 planner run', () => {
  withTmpDataDir(() => {
    const { store, taskId, runId } = setupPlanning('medium');
    assert.equal(store.getTask(taskId)!.status, 'speccing');
    appendOutbox(runId, { type: 'started', pid: 4243, worktree: '/r', branch: 'n/a', model: 'opus' });
    appendOutbox(runId, { type: 'spec-result', spec: '## 改动\n改 a.ts', transcriptRef: '/p.jsonl', sessionId: 's' });
    appendOutbox(runId, { type: 'done' });
    ingestRun(store, noopLog, runId);
    const t = store.getTask(taskId)!;
    assert.equal(t.status, 'spec_review');
    assert.equal(t.spec, '## 改动\n改 a.ts');
    assert.equal(store.getRun(runId)!.status, 'succeeded');
    store.close();
  });
});

test('ingest planner-decompose：setPlan + 建子任务 + 转 plan_review', () => {
  withTmpDataDir(() => {
    const { store, taskId, runId } = setupPlanning('hard');
    assert.equal(store.getTask(taskId)!.status, 'analyzing');
    appendOutbox(runId, {
      type: 'decompose-result', plan: '拆成两步',
      subtasks: [{ title: '建骨架', complexity: 'easy' }, { title: '实现', complexity: 'medium' }],
      transcriptRef: null, sessionId: null,
    });
    appendOutbox(runId, { type: 'done' });
    ingestRun(store, noopLog, runId);
    const t = store.getTask(taskId)!;
    assert.equal(t.status, 'plan_review');
    assert.equal(t.plan, '拆成两步');
    const kids = store.listTasks(t.projectId).filter((x) => x.parentId === taskId);
    assert.equal(kids.length, 2, '建了 2 个子任务');
    assert.deepEqual(kids.map((k) => k.complexity).sort(), ['easy', 'medium']);
    assert.equal(store.getRun(runId)!.status, 'succeeded');
    store.close();
  });
});

test('ingest planner failed：走 failPlanAttempt（退避，任务留 speccing，不进 failed）', () => {
  withTmpDataDir(() => {
    const { store, taskId, runId } = setupPlanning('medium');
    appendOutbox(runId, { type: 'failed', error: 'planner 崩了', transcriptRef: null, sessionId: null });
    appendOutbox(runId, { type: 'done' });
    ingestRun(store, noopLog, runId);
    const t = store.getTask(taskId)!;
    assert.equal(t.status, 'speccing', '默认 maxRetries=2，第一次失败留 speccing 退避重试');
    assert.ok(t.nextEligibleAt, '设了退避 nextEligibleAt');
    assert.equal(store.getRun(runId)!.status, 'failed');
    store.close();
  });
});