maestro/test/pipeline.test.ts

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { runPipeline, parseDecompose, type PipelineDeps } from '../src/executor/pipeline.js';
import type { JobSpec, OutboxPayload } from '../src/executor/protocol.js';
import type { Project, Task } from '../src/model/types.js';
import type { RunnerResult } from '../src/executor/runner.js';
import type { ReviewResult } from '../src/executor/reviewer.js';

// ───────────────────────── 固定 fixtures ─────────────────────────

const TASK_ID = 'task-pipe-1';
const RUN_ID = 'run-pipe-1';
const FAKE_DIR = `/tmp/fake-wt/${TASK_ID}`;
const FAKE_BRANCH = `maestro/${TASK_ID}`;

const fakeProject: Project = {
  id: 'proj-1', name: 'pipe', repoPath: '/tmp/pipe-repo', defaultBranch: 'main',
  verifyCmd: null, autonomy: 'auto-easy', model: null, concurrency: 1, maxRetries: 2,
  timeoutMs: 1_800_000, status: 'active', logo: null, sortOrder: 0,
  createdAt: '2024-01-01T00:00:00.000Z', lastSyncAt: null,
};

const fakeTask: Task = {
  id: TASK_ID, projectId: 'proj-1', parentId: null, depth: 1, title: 'tweak',
  complexity: 'easy', status: 'queued', priority: 0, deps: [],
  plan: null, spec: null, operations: '在 README.md 追加一行', approvals: [],
  result: null, assignee: 'agent', retryBaseline: 0, nextEligibleAt: null,
  createdAt: '2024-01-01T00:00:00.000Z', updatedAt: '2024-01-01T00:00:00.000Z',
};

const fakeJob: JobSpec = {
  runId: RUN_ID, task: fakeTask, project: fakeProject,
  worktreeDir: FAKE_DIR, branch: FAKE_BRANCH,
};

const okRun: RunnerResult = {
  ok: true, transcriptRef: '/tmp/fake.jsonl', sessionId: 'sess-exec-1',
  finalText: '执行自述：改了 README',
};
const okReview: ReviewResult = {
  summary: '## 做了什么\nmock 复审通过', verdict: 'approve',
  transcriptRef: '/tmp/fake-review.jsonl', sessionId: 'sess-review-1',
};
const okSecurity: ReviewResult = {
  summary: '## 安全审计\nmock 审计通过', verdict: 'approve',
  transcriptRef: '/tmp/fake-security.jsonl', sessionId: 'sess-security-1',
};

/** 全 mock 依赖（不真起 CC、不动 git）：可按用例覆盖。 */
function mockDeps(overrides: Partial<PipelineDeps> = {}): PipelineDeps {
  return {
    createWorktree: async (_repo, taskId) => ({ dir: `/tmp/fake-wt/${taskId}`, branch: `maestro/${taskId}` }),
    worktreeDiff: async () => ({ diffSummary: ' README.md | 1 +', commits: ['abc1234 hello maestro'] }),
    verify: async () => ({ ok: true, exitCode: 0, logRef: null }),
    runTask: async () => okRun,
    reviewCode: async () => okReview,
    reviewSecurity: async () => okSecurity,
    runPlanner: async () => ({ ok: true, transcriptRef: '/tmp/plan.jsonl', sessionId: 'sess-plan', finalText: '方案正文' }),
    runConflict: async () => ({ ok: true, transcriptRef: null, sessionId: null, finalText: null }),
    syncMain: async () => null,  // no-op：测试环境无真实 git worktree，跳过同步
    ...overrides,
  };
}

/** 收集 emit 出的 payload 序列；返回收集器与 emit 函数。 */
function collector(): { emitted: OutboxPayload[]; emit: (p: OutboxPayload) => void } {
  const emitted: OutboxPayload[] = [];
  return { emitted, emit: (p) => { emitted.push(p); } };
}

/** 取某类型的（首条）payload。 */
function find<T extends OutboxPayload['type']>(
  emitted: OutboxPayload[], type: T,
): Extract<OutboxPayload, { type: T }> | undefined {
  return emitted.find((p) => p.type === type) as Extract<OutboxPayload, { type: T }> | undefined;
}

// ───────────────────────── 用例 ─────────────────────────

test('成功路径：emit result（四要素正确）+ 末尾 done，不含 failed', async () => {
  const { emitted, emit } = collector();
  await runPipeline(fakeJob, mockDeps(), emit);

  // 不含 failed
  assert.equal(find(emitted, 'failed'), undefined, '成功路径不应 emit failed');

  // 末尾是 done
  assert.equal(emitted.at(-1)?.type, 'done', '最后一条应为 done');

  // result 四要素
  const result = find(emitted, 'result');
  assert.ok(result, '应 emit result');
  assert.equal(result.branch, FAKE_BRANCH);
  assert.equal(result.worktree, FAKE_DIR);
  assert.equal(result.diffSummary, ' README.md | 1 +');
  assert.deepEqual(result.commits, ['abc1234 hello maestro']);
  // executor 透传 runner 的 transcriptRef/sessionId
  assert.deepEqual(result.executor, { transcriptRef: '/tmp/fake.jsonl', sessionId: 'sess-exec-1' });
  // code review report（ReviewResult → ReviewReport：丢 sessionId）
  assert.deepEqual(result.code, {
    summary: '## 做了什么\nmock 复审通过', verdict: 'approve', transcriptRef: '/tmp/fake-review.jsonl',
  });
  // security report
  assert.deepEqual(result.security, {
    summary: '## 安全审计\nmock 审计通过', verdict: 'approve', transcriptRef: '/tmp/fake-security.jsonl',
  });

  // done 恰好一条
  assert.equal(emitted.filter((p) => p.type === 'done').length, 1);
});

test('双复审收到的执行者自述＝runner.finalText', async () => {
  let codeReport: string | null = null;
  let secReport: string | null = null;
  const { emit } = collector();
  await runPipeline(fakeJob, mockDeps({
    reviewCode: async (_t, _p, _wt, _runId, report) => { codeReport = report; return okReview; },
    reviewSecurity: async (_t, _p, _wt, _runId, report) => { secReport = report; return okSecurity; },
  }), emit);
  assert.equal(codeReport, '执行自述：改了 README');
  assert.equal(secReport, '执行自述：改了 README');
});

test('executor 失败：emit failed（携带 error/transcriptRef/sessionId）+ done，无 result', async () => {
  const { emitted, emit } = collector();
  await runPipeline(fakeJob, mockDeps({
    runTask: async () => ({ ok: false, transcriptRef: '/tmp/exec-fail.jsonl', sessionId: 'sess-x', error: 'boom 执行炸了' }),
  }), emit);

  assert.equal(find(emitted, 'result'), undefined, 'executor 失败不应 emit result');
  const failed = find(emitted, 'failed');
  assert.ok(failed, '应 emit failed');
  assert.equal(failed.error, 'boom 执行炸了');
  assert.equal(failed.transcriptRef, '/tmp/exec-fail.jsonl');
  assert.equal(failed.sessionId, 'sess-x');
  assert.equal(emitted.at(-1)?.type, 'done', '末尾应为 done');
});

test('verify 失败：runner ok 但 verify !ok → emit failed（沿用 executor transcript）+ done，无 result', async () => {
  const { emitted, emit } = collector();
  await runPipeline(fakeJob, mockDeps({
    verify: async () => ({ ok: false, exitCode: 1, logRef: '/tmp/v.log', error: 'verify 失败（exit 1）' }),
  }), emit);

  assert.equal(find(emitted, 'result'), undefined, 'verify 失败不应 emit result');
  const failed = find(emitted, 'failed');
  assert.ok(failed, '应 emit failed');
  assert.equal(failed.error, 'verify 失败（exit 1）');
  // transcriptRef/sessionId 沿用 executor 的（okRun）
  assert.equal(failed.transcriptRef, '/tmp/fake.jsonl');
  assert.equal(failed.sessionId, 'sess-exec-1');
  assert.equal(emitted.at(-1)?.type, 'done', '末尾应为 done');
});

test('checks 闸失败：verify 过但分项检查 !ok → emit failed（报哪项挂）+ done，无 result，不复审', async () => {
  const { emitted, emit } = collector();
  let reviewed = false;
  await runPipeline(fakeJob, mockDeps({
    checks: async () => ({ ok: false, failed: 'typecheck', logRef: '/tmp/c.log', error: 'typecheck 检查失败（exit 1）：tsc' }),
    reviewCode: async () => { reviewed = true; return okReview; },
  }), emit);

  assert.equal(find(emitted, 'result'), undefined, 'checks 失败不应 emit result');
  assert.equal(reviewed, false, 'checks 失败不应进入复审');
  const failed = find(emitted, 'failed');
  assert.ok(failed, '应 emit failed');
  assert.match(failed.error!, /typecheck 检查失败/);
  assert.equal(emitted.at(-1)?.type, 'done', '末尾应为 done');
});

test('diff 体量闸失败：emit failed（体量超阈值）+ done，无 result', async () => {
  const { emitted, emit } = collector();
  await runPipeline(fakeJob, mockDeps({
    diffGate: async () => ({ ok: false, files: 99, lines: 9999, error: '改动体量超阈值（文件数 99 > 60）：任务可能过大，建议拆解后重做' }),
  }), emit);

  assert.equal(find(emitted, 'result'), undefined, 'diff 闸失败不应 emit result');
  const failed = find(emitted, 'failed');
  assert.ok(failed, '应 emit failed');
  assert.match(failed.error!, /体量超阈值/);
  assert.equal(emitted.at(-1)?.type, 'done', '末尾应为 done');
});

test('复审抛错不挡：reviewCode 抛错 → result 仍 emit，code.verdict=null 且 summary 含「自动复审失败」', async () => {
  const { emitted, emit } = collector();
  await runPipeline(fakeJob, mockDeps({
    reviewCode: async () => { throw new Error('复审 CC 崩了'); },
  }), emit);

  const result = find(emitted, 'result');
  assert.ok(result, '复审抛错仍应 emit result');
  assert.equal(result.code.verdict, null);
  assert.equal(result.code.transcriptRef, null);
  assert.match(result.code.summary ?? '', /自动复审失败/);
  assert.match(result.code.summary ?? '', /复审 CC 崩了/);
  // 另一个复审不受影响
  assert.equal(result.security.verdict, 'approve');
  assert.equal(result.security.summary, '## 安全审计\nmock 审计通过');
  assert.equal(find(emitted, 'failed'), undefined, '复审失败不应升级为 failed');
  assert.equal(emitted.at(-1)?.type, 'done');
});

test('安全审计抛错不挡：security.verdict=null、summary 含「自动复审失败」，code review 不受影响', async () => {
  const { emitted, emit } = collector();
  await runPipeline(fakeJob, mockDeps({
    reviewSecurity: async () => { throw new Error('审计 CC 崩了'); },
  }), emit);

  const result = find(emitted, 'result');
  assert.ok(result);
  assert.equal(result.security.verdict, null);
  assert.match(result.security.summary ?? '', /自动复审失败：审计 CC 崩了/);
  assert.equal(result.code.verdict, 'approve');
  assert.equal(emitted.at(-1)?.type, 'done');
});

// ───────────────────────── planner 分支 ─────────────────────────

const medJob = (over: Partial<JobSpec> = {}): JobSpec => ({ ...fakeJob, runKind: 'planner-spec', task: { ...fakeTask, complexity: 'medium', status: 'speccing' }, ...over });
const hardJob = (over: Partial<JobSpec> = {}): JobSpec => ({ ...fakeJob, runKind: 'planner-decompose', task: { ...fakeTask, complexity: 'hard', status: 'analyzing' }, ...over });

test('planner-spec：emit spec-result（方案=CC 文本）+ done，不复审/不建 worktree', async () => {
  const { emitted, emit } = collector();
  let worktreeCalled = false;
  await runPipeline(medJob(), mockDeps({
    createWorktree: async () => { worktreeCalled = true; return { dir: 'x', branch: 'y' }; },
    runPlanner: async () => ({ ok: true, transcriptRef: '/tmp/p.jsonl', sessionId: 's', finalText: '## 改动\n改 a.ts' }),
  }), emit);
  assert.equal(worktreeCalled, false, 'planner 不建 worktree');
  const sr = find(emitted, 'spec-result');
  assert.ok(sr, '应 emit spec-result');
  assert.equal(sr.spec, '## 改动\n改 a.ts');
  assert.equal(find(emitted, 'result'), undefined);
  assert.equal(find(emitted, 'failed'), undefined);
  assert.equal(emitted.at(-1)?.type, 'done');
});

test('planner-decompose：解析末尾 fenced JSON → emit decompose-result + done', async () => {
  const { emitted, emit } = collector();
  const txt = '分析：拆成两步。\n```json\n{"plan":"先骨架再实现","subtasks":[{"title":"建骨架","complexity":"easy"},{"title":"实现逻辑","complexity":"medium"}]}\n```';
  await runPipeline(hardJob(), mockDeps({ runPlanner: async () => ({ ok: true, transcriptRef: null, sessionId: null, finalText: txt }) }), emit);
  const dr = find(emitted, 'decompose-result');
  assert.ok(dr, '应 emit decompose-result');
  assert.equal(dr.plan, '先骨架再实现');
  assert.equal(dr.subtasks.length, 2);
  assert.deepEqual(dr.subtasks.map((s) => s.complexity), ['easy', 'medium']);
  assert.equal(emitted.at(-1)?.type, 'done');
});

test('planner-decompose：输出无合法 JSON → emit failed', async () => {
  const { emitted, emit } = collector();
  await runPipeline(hardJob(), mockDeps({ runPlanner: async () => ({ ok: true, transcriptRef: null, sessionId: null, finalText: '我忘了输出 JSON' }) }), emit);
  assert.ok(find(emitted, 'failed'), '解析失败应 emit failed');
  assert.equal(find(emitted, 'decompose-result'), undefined);
  assert.equal(emitted.at(-1)?.type, 'done');
});

test('planner：CC 失败 → emit failed', async () => {
  const { emitted, emit } = collector();
  await runPipeline(medJob(), mockDeps({ runPlanner: async () => ({ ok: false, transcriptRef: null, sessionId: null, finalText: null, error: 'CC 崩了' }) }), emit);
  const f = find(emitted, 'failed');
  assert.ok(f);
  assert.match(f.error, /CC 崩了/);
  assert.equal(emitted.at(-1)?.type, 'done');
});

test('规则13 分歧重评估：reevalGate 撞声明范围 → emit needs-reeval + done，不执行/不复审/无 failed', async () => {
  const { emitted, emit } = collector();
  const job: JobSpec = { ...fakeJob, task: { ...fakeTask, scopeFiles: ['src/**'] } };
  let ranTask = false;
  await runPipeline(job, mockDeps({
    reevalGate: async () => ({ overlap: ['src/a.ts', 'src/b.ts'] }),
    runTask: async () => { ranTask = true; return okRun; },
  }), emit);

  const re = find(emitted, 'needs-reeval');
  assert.ok(re, '撞声明范围应 emit needs-reeval');
  assert.deepEqual(re.files, ['src/a.ts', 'src/b.ts']);
  assert.match(re.reason, /原方案可能已过时/);
  assert.equal(ranTask, false, '重评估时不应执行任务');
  assert.equal(find(emitted, 'result'), undefined, '不应 emit result');
  assert.equal(find(emitted, 'failed'), undefined, '重评估不是失败，不应 emit failed');
  assert.equal(emitted.at(-1)?.type, 'done', '末尾应为 done');
});

test('规则13 分歧重评估：无重叠 → 正常往下执行（emit result）', async () => {
  const { emitted, emit } = collector();
  const job: JobSpec = { ...fakeJob, task: { ...fakeTask, scopeFiles: ['src/**'] } };
  await runPipeline(job, mockDeps({ reevalGate: async () => ({ overlap: [] }) }), emit);
  assert.equal(find(emitted, 'needs-reeval'), undefined, '无重叠不应触发重评估');
  assert.ok(find(emitted, 'result'), '应正常执行到 result');
});

test('parseDecompose：取最后一个 json 块；非法/缺失 → null', () => {
  assert.equal(parseDecompose('no json here'), null);
  assert.equal(parseDecompose('```json\n{"subtasks":[]}\n```'), null, '空 subtasks → null');
  const ok = parseDecompose('前文\n```json\n{"subtasks":[{"title":"a","complexity":"hard"},{"title":"","complexity":"easy"},{"title":"b","complexity":"x"}]}\n```');
  assert.ok(ok);
  assert.equal(ok.subtasks.length, 1, '过滤空标题与非法复杂度');
  assert.equal(ok.subtasks[0].title, 'a');
});

test('parseDecompose：解析 files（声明文件范围），去空白/空项；缺省不带 files', () => {
  const r = parseDecompose('```json\n{"plan":"p","subtasks":[{"title":"a","complexity":"easy","files":[" src/** ","","lib/x.ts"]},{"title":"b","complexity":"easy"}]}\n```');
  assert.ok(r);
  assert.deepEqual(r.subtasks[0].files, ['src/**', 'lib/x.ts']);
  assert.equal(r.subtasks[1].files, undefined, '无 files 字段 → 不带（不限范围）');
});