diff --git a/src/executor/pipeline.ts b/src/executor/pipeline.ts new file mode 100644 index 0000000..7ff89a8 --- /dev/null +++ b/src/executor/pipeline.ts @@ -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; + worktreeDiff: (repoPath: string, dir: string, branch: string, baseBranch: string) => Promise; + 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 { + 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 { + // 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' }); +} diff --git a/src/executor/worker.ts b/src/executor/worker.ts new file mode 100644 index 0000000..0d9129d --- /dev/null +++ b/src/executor/worker.ts @@ -0,0 +1,93 @@ +// Phase 3 多进程执行:worker 进程【入口】。 +// +// node worker.js (或 tsx worker.ts ) +// +// daemon 在 spawn 前写好 runs//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 { + const runId = process.argv[2]; + if (!runId) { + // 没 runId 没法定位 job.json / outbox,无处汇报——只能直接退出(非 0 让父进程可感知)。 + process.stderr.write('worker: 缺少 runId 参数(用法:worker )\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); +}); diff --git a/test/pipeline.test.ts b/test/pipeline.test.ts new file mode 100644 index 0000000..304bd84 --- /dev/null +++ b/test/pipeline.test.ts @@ -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 { + 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( + emitted: OutboxPayload[], type: T, +): Extract | undefined { + return emitted.find((p) => p.type === type) as Extract | 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'); +});