phase3(worker): pipeline.ts 纯执行管线 + worker.ts 进程入口（不碰 DB，经 outbox 文件汇报）

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 12:56:50 +08:00
parent 007dffac39
commit d8f8b5da1e
3 changed files with 396 additions and 0 deletions
@@ -0,0 +1,120 @@
+// Phase 3 多进程执行：worker 侧【纯执行管线】。
+//
+// 这是旧 `orchestrator.executeTask` 函数体「劈出来」的执行半边——只负责跑
+// worktree→执行→verify→双复审→产出，并把每一步的进度/结果经 `emit` 写进 outbox 文件。
+// 它【完全不碰 DB】（连读都不碰），也【不做任何重试 / needs_attention 决策】：
+//   - 失败（executor 报错 / verify 不过）→ emit failed，然后 emit done，结束（不抛错）。
+//   - 成功 → emit result（四要素），然后 emit done。
+// 重试次数、退避、转终态等判断全部归 daemon（它读 outbox 后决策）。
+//
+// daemon 侧的 store.* 调用与本模块 emit 的对应关系（仅为说明，本模块不引用 store）：
+//   store.transition(executing) / startRun(executor)  → emit started（由 worker 在入口发，不在此）
+//   runner 失败 / verify 失败 finishRun(failed)        → emit failed
+//   双复审 + setResult(四字段) + transition(exec_review) → emit result
+//   （worker 退出）                                      → emit done
+
+import type { JobSpec, OutboxPayload, ReviewReport } from './protocol.js';
+import { createWorktree, worktreeDiff, type WorktreeDiff, type WorktreeInfo } from './worktree.js';
+import { runTask, type RunnerFn } from './runner.js';
+import { runVerify, type VerifyFn } from './verify.js';
+import { reviewCode, reviewSecurity, type ReviewerFn } from './reviewer.js';
+
+/** pipeline 的依赖注入面（便于单测 mock）；默认值取真实现。 */
+export interface PipelineDeps {
+  createWorktree: (repoPath: string, taskId: string, baseBranch: string) => Promise<WorktreeInfo>;
+  worktreeDiff: (repoPath: string, dir: string, branch: string, baseBranch: string) => Promise<WorktreeDiff>;
+  runTask: RunnerFn;
+  verify: VerifyFn;
+  reviewCode: ReviewerFn;     // code review（daemon 落成 kind=reviewer 的 run）
+  reviewSecurity: ReviewerFn; // 安全审计（daemon 落成 kind=security 的 run）
+}
+
+/** 真实依赖（worker 生产用）。 */
+export const realDeps: PipelineDeps = {
+  createWorktree,
+  worktreeDiff,
+  runTask,
+  verify: runVerify,
+  reviewCode,
+  reviewSecurity,
+};
+
+/** emit：把一条 OutboxPayload 交给 outbox（worker 传 appendOutbox 包装，测试传收集器）。 */
+export type Emit = (payload: OutboxPayload) => void;
+
+/**
+ * 跑一个独立复审（code / security），复刻旧 `runOneReview` 的兜底：
+ * 任何失败折叠成 `{summary:'自动复审失败：'+msg, verdict:null, transcriptRef:null}`，不抛错、不挡任务。
+ * 成功时把 ReviewResult 映射成 ReviewReport（丢弃 sessionId——result 记录里复审只留 transcriptRef）。
+ */
+async function runOneReview(
+  fn: ReviewerFn,
+  job: JobSpec,
+  wt: WorktreeInfo,
+  reviewRunId: string,
+  executorReport: string,
+): Promise<ReviewReport> {
+  try {
+    const rv = await fn(job.task, job.project, wt, reviewRunId, executorReport);
+    return { summary: rv.summary, verdict: rv.verdict, transcriptRef: rv.transcriptRef };
+  } catch (e) {
+    return { summary: `自动复审失败：${(e as Error).message}`, verdict: null, transcriptRef: null };
+  }
+}
+
+/**
+ * 纯执行管线。复刻旧 in-process 流程的【执行与产出】，但所有 store.* 改成 emit(outbox)：
+ *  1. createWorktree
+ *  2. runTask；!ok → emit failed{error,transcriptRef,sessionId} + done，返回（不抛）
+ *  3. runVerify；!ok → emit failed{error,rr.transcriptRef,rr.sessionId} + done，返回
+ *  4. worktreeDiff
+ *  5. 双复审（顺序，任一失败不挡，见 runOneReview 兜底）
+ *  6. emit result（branch/worktree/diffSummary/commits/executor/code/security）+ done
+ *
+ * 不抛错（除非 emit 自身抛——那由 worker 顶层兜底）。全程总会 emit 一条终态（failed|result）后再 emit done。
+ */
+export async function runPipeline(job: JobSpec, deps: PipelineDeps = realDeps, emit: Emit): Promise<void> {
+  // 1. 建 worktree（其 dir/branch 应与 job.worktreeDir/branch 一致；以 deps 真建的为准）
+  const wt = await deps.createWorktree(job.project.repoPath, job.task.id, job.project.defaultBranch);
+
+  // 2. 执行
+  emit({ type: 'phase', phase: 'executing' });
+  const rr = await deps.runTask(job.task, job.project, wt, job.runId);
+  if (!rr.ok) {
+    emit({ type: 'failed', error: rr.error ?? '执行失败', transcriptRef: rr.transcriptRef, sessionId: rr.sessionId });
+    emit({ type: 'done' });
+    return;
+  }
+
+  // 3. 校验
+  emit({ type: 'phase', phase: 'verifying' });
+  const vr = await deps.verify(job.project, wt.dir, job.runId);
+  if (!vr.ok) {
+    // transcriptRef/sessionId 沿用 executor 的（verify 自身无转录）
+    emit({ type: 'failed', error: vr.error ?? 'verify 失败', transcriptRef: rr.transcriptRef, sessionId: rr.sessionId });
+    emit({ type: 'done' });
+    return;
+  }
+
+  // 4. diff
+  const diff = await deps.worktreeDiff(job.project.repoPath, wt.dir, wt.branch, job.project.defaultBranch);
+
+  // 5. 双复审（顺序；任一失败不挡）。runId 派生子 id，与旧实现一致（${runId}.review / ${runId}.security）
+  emit({ type: 'phase', phase: 'reviewing' });
+  const report = rr.finalText ?? '';
+  const code = await runOneReview(deps.reviewCode, job, wt, `${job.runId}.review`, report);
+  const security = await runOneReview(deps.reviewSecurity, job, wt, `${job.runId}.security`, report);
+
+  // 6. 成功终态
+  emit({
+    type: 'result',
+    branch: wt.branch,
+    worktree: wt.dir,
+    diffSummary: diff.diffSummary,
+    commits: diff.commits,
+    executor: { transcriptRef: rr.transcriptRef, sessionId: rr.sessionId },
+    code,
+    security,
+  });
+  emit({ type: 'done' });
+}
@@ -0,0 +1,93 @@
+// Phase 3 多进程执行：worker 进程【入口】。
+//
+//   node worker.js <runId>   （或 tsx worker.ts <runId>）
+//
+// daemon 在 spawn 前写好 runs/<runId>/job.json，本进程读它即可执行，【完全不碰 DB】。
+// 一切进度/结果只经 protocol 的文件协议汇报：
+//   - 起手 emit started（pid/worktree/branch/model）
+//   - 周期 touchHeartbeat（daemon 据 mtime 判活）
+//   - runPipeline 内部 emit failed|result + done（终态）
+//   - 意外异常 / 收到 SIGTERM → 兜底 emit failed + done，然后退出
+//
+// 顶层【不 import 任何 store/db】——worker 与 DB 完全隔离。
+
+import {
+  readJobSpec,
+  appendOutbox,
+  touchHeartbeat,
+  HEARTBEAT_INTERVAL_MS,
+  type OutboxPayload,
+} from './protocol.js';
+import { pickModel } from './models.js';
+import { runPipeline, realDeps } from './pipeline.js';
+
+async function main(): Promise<void> {
+  const runId = process.argv[2];
+  if (!runId) {
+    // 没 runId 没法定位 job.json / outbox，无处汇报——只能直接退出（非 0 让父进程可感知）。
+    process.stderr.write('worker: 缺少 runId 参数（用法：worker <runId>）\n');
+    process.exit(2);
+  }
+
+  const emit = (p: OutboxPayload): void => { appendOutbox(runId, p); };
+  const job = readJobSpec(runId);
+
+  // 起手汇报：pid / worktree / branch / 实际执行模型
+  emit({
+    type: 'started',
+    pid: process.pid,
+    worktree: job.worktreeDir,
+    branch: job.branch,
+    model: pickModel(job.task, job.project, 'executor'),
+  });
+
+  // 心跳：立即一次 + 周期刷（结束时清掉）
+  touchHeartbeat(runId);
+  const hb = setInterval(() => touchHeartbeat(runId), HEARTBEAT_INTERVAL_MS);
+  // 心跳定时器不应拖住事件循环退出（正常路径我们显式 exit，这里只是兜底）
+  hb.unref?.();
+
+  // SIGTERM＝daemon 取消/超时：兜底报失败终态后干净退出（exit 0：已自报终态，不算崩溃）
+  let signalled = false;
+  process.on('SIGTERM', () => {
+    if (signalled) return;
+    signalled = true;
+    clearInterval(hb);
+    try {
+      emit({ type: 'failed', error: 'worker 收到 SIGTERM（取消/超时）', transcriptRef: null, sessionId: null });
+      emit({ type: 'done' });
+    } catch { /* 汇报失败也要退出，别卡死 */ }
+    process.exit(0);
+  });
+
+  try {
+    // runPipeline 内部已 emit failed|result + done；正常路径这里不再补发终态
+    await runPipeline(job, realDeps, emit);
+  } catch (e) {
+    // 意外异常（pipeline 之外或 emit 抛错等）：兜底补一条 failed + done
+    if (!signalled) {
+      try {
+        emit({ type: 'failed', error: (e as Error).message, transcriptRef: null, sessionId: null });
+        emit({ type: 'done' });
+      } catch { /* 已尽力汇报 */ }
+    }
+  } finally {
+    clearInterval(hb);
+  }
+
+  if (!signalled) process.exit(0);
+}
+
+main().catch((e) => {
+  // main 自身（如 readJobSpec 抛错）兜底：尽量写一条 failed，但若连 runId 都没有就只能干退。
+  const runId = process.argv[2];
+  if (runId) {
+    try {
+      appendOutbox(runId, { type: 'failed', error: (e as Error).message, transcriptRef: null, sessionId: null });
+      appendOutbox(runId, { type: 'done' });
+    } catch { /* 无处可报 */ }
+  } else {
+    process.stderr.write(`worker 致命错误：${(e as Error).message}\n`);
+  }
+  process.exit(1);
+});
@@ -0,0 +1,183 @@
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import { runPipeline, 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,
+    ...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('复审抛错不挡：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');
+});