wangjia/maestro
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(pipeline): 并行复审+sync-main+conflict-pipeline

Browse Source 
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
wangjia
2026-06-14 12:23:57 +08:00
parent 76d11a8307
commit ae9ff7a469
1 changed files with 122 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/executor/pipeline.ts
+122 -6
View File
@@ -14,8 +14,8 @@
// （worker 退出） → emit done
import type { JobSpec, OutboxPayload, ReviewReport, DecomposeResult } from './protocol.js';
import { createWorktree, worktreeDiff, type WorktreeDiff, type WorktreeInfo } from './worktree.js';
import { runTask, runPlanner, type RunnerFn, type PlannerFn } from './runner.js';
import { createWorktree, worktreeDiff, git, type WorktreeDiff, type WorktreeInfo } from './worktree.js';
import { runTask, runPlanner, runConflict, type RunnerFn, type PlannerFn } from './runner.js';
import { runVerify, type VerifyFn } from './verify.js';
import { reviewCode, reviewSecurity, type ReviewerFn } from './reviewer.js';
@@ -28,6 +28,7 @@ export interface PipelineDeps {
reviewCode: ReviewerFn; // code review（daemon 落成 kind=reviewer 的 run）
reviewSecurity: ReviewerFn; // 安全审计（daemon 落成 kind=security 的 run）
runPlanner: PlannerFn; // planner（拆解 Hard / 写方案 Medium，只读跑 CC）
runConflict: typeof runConflict; // 解冲突（conflict run）
}
/** 真实依赖（worker 生产用）。 */
@@ -39,6 +40,7 @@ export const realDeps: PipelineDeps = {
reviewCode,
reviewSecurity,
runPlanner,
runConflict,
};
/** emit：把一条 OutboxPayload 交给 outbox（worker 传 appendOutbox 包装，测试传收集器）。 */
@@ -80,9 +82,38 @@ export async function runPipeline(job: JobSpec, deps: PipelineDeps = realDeps, e
if (job.runKind === 'planner-spec' || job.runKind === 'planner-decompose') {
return runPlannerPipeline(job, deps, emit);
}
// conflict 走解冲突管线
if (job.runKind === 'conflict') {
return runConflictPipeline(job, deps, emit);
}
// 1. 建 worktree（其 dir/branch 应与 job.worktreeDir/branch 一致；以 deps 真建的为准）
const wt = await deps.createWorktree(job.project.repoPath, job.task.id, job.project.defaultBranch);
// sync main：把 defaultBranch 最新代码 merge 进 worktree（排队期间 main 可能已前移）
try {
await git(wt.dir, ['fetch', 'origin', job.project.defaultBranch]);
} catch { /* 无远端时跳过 fetch，直接用本地 */ }
try {
await git(wt.dir, [
'-c', 'user.name=maestro', '-c', 'user.email=maestro@local',
'merge', '--no-edit', `origin/${job.project.defaultBranch}`,
]);
} catch {
// 尝试用本地分支
try {
await git(wt.dir, [
'-c', 'user.name=maestro', '-c', 'user.email=maestro@local',
'merge', '--no-edit', job.project.defaultBranch,
]);
} catch (e) {
// sync main 失败（冲突或其他）→ abort 并报失败，让 daemon 转 needs_attention
await git(wt.dir, ['merge', '--abort']).catch(() => undefined);
emit({ type: 'failed', error: `执行前同步 ${job.project.defaultBranch} 失败，建议重新评估方案：${(e as Error).message}`, transcriptRef: null, sessionId: null });
emit({ type: 'done' });
return;
}
}
// 2. 执行
emit({ type: 'phase', phase: 'executing' });
const rr = await deps.runTask(job.task, job.project, wt, job.runId);
@@ -105,11 +136,13 @@ export async function runPipeline(job: JobSpec, deps: PipelineDeps = realDeps, e
// 4. diff
const diff = await deps.worktreeDiff(job.project.repoPath, wt.dir, wt.branch, job.project.defaultBranch);
// 5. 双复审（顺序；任一失败不挡）。runId 派生子 id，与旧实现一致（${runId}.review / ${runId}.security）
// 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);
const [code, security] = await Promise.all([
runOneReview(deps.reviewCode, job, wt, `${job.runId}.review`, report),
runOneReview(deps.reviewSecurity, job, wt, `${job.runId}.security`, report),
]);
// 6. 成功终态
emit({
@@ -155,6 +188,79 @@ async function runPlannerPipeline(job: JobSpec, deps: PipelineDeps, emit: Emit):
emit({ type: 'done' });
}
/**
* 解冲突管线（conflict run）：
* 1. 建 worktree（从 defaultBranch 建，和普通 executor 一样）
* 2. 在 worktree 内执行 git merge <原分支> 触发冲突态
* 3. 交 CC 解冲突（runConflict）
* 4. 双复审（并行，同 executor）
* 5. emit result
*/
async function runConflictPipeline(job: JobSpec, deps: PipelineDeps, emit: Emit): Promise<void> {
// 1. 建 worktree
const wt = await deps.createWorktree(job.project.repoPath, job.task.id, job.project.defaultBranch);
// 2. 在 worktree 内触发 git merge <原任务分支> 制造冲突态
// job.task.result.branch 是原任务分支（由 server.ts 在建 conflict run 时写入 job）
const originalBranch = job.task.result?.branch ?? '';
if (!originalBranch) {
emit({ type: 'failed', error: '解冲突任务缺少原任务分支信息（result.branch）', transcriptRef: null, sessionId: null });
emit({ type: 'done' });
return;
}
emit({ type: 'phase', phase: 'merging' });
try {
await git(wt.dir, [
'-c', 'user.name=maestro', '-c', 'user.email=maestro@local',
'merge', '--no-edit', originalBranch,
]);
// merge 成功（无冲突）→ 直接走正常双复审流程
} catch {
// merge 产生冲突，正是预期状态，继续让 CC 解
}
// 3. 收集冲突文件列表
let conflictFiles: string[] = [];
try {
const out = await git(wt.dir, ['diff', '--name-only', '--diff-filter=U']);
conflictFiles = out.trim().split('\n').filter(Boolean);
} catch { /* 忽略 */ }
// 4. CC 解冲突
emit({ type: 'phase', phase: 'resolving' });
const rr = await deps.runConflict(job.task, job.project, wt, job.runId, conflictFiles);
if (!rr.ok) {
emit({ type: 'failed', error: rr.error ?? '解冲突失败', transcriptRef: rr.transcriptRef, sessionId: rr.sessionId });
emit({ type: 'done' });
return;
}
// 5. diff
const diff = await deps.worktreeDiff(job.project.repoPath, wt.dir, wt.branch, job.project.defaultBranch);
// 6. 双复审（并行）
emit({ type: 'phase', phase: 'reviewing' });
const report = rr.finalText ?? '';
const [code, security] = await Promise.all([
runOneReview(deps.reviewCode, job, wt, `${job.runId}.review`, report),
runOneReview(deps.reviewSecurity, job, wt, `${job.runId}.security`, report),
]);
// 7. 成功终态
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' });
}
/** 从 CC 文本里抽取拆解结果：优先取最后一个 ```json 块，回退到任意 ``` 块，再回退到整段当 JSON。非法 → null。 */
export function parseDecompose(text: string): DecomposeResult | null {
const blocks = [...text.matchAll(/```(?:json)?\s*([\s\S]*?)```/gi)].map((m) => m[1].trim());
@@ -167,7 +273,17 @@ export function parseDecompose(text: string): DecomposeResult | null {
.filter((s): s is { title: string; complexity: 'easy' | 'medium' | 'hard' } =>
!!s && typeof (s as { title?: unknown }).title === 'string' &&
['easy', 'medium', 'hard'].includes((s as { complexity?: unknown }).complexity as string))
.map((s) => ({ title: String(s.title).trim(), complexity: s.complexity }))
.map((s) => {
const raw = s as unknown as { priority?: unknown; deps?: unknown };
return {
title: String(s.title).trim(),
complexity: s.complexity,
priority: typeof raw.priority === 'number' ? raw.priority : 1,
deps: Array.isArray(raw.deps)
? (raw.deps as unknown[]).filter((d): d is number => typeof d === 'number')
: [] as number[],
};
})
.filter((s) => s.title.length > 0);
if (subtasks.length === 0) continue;
return { plan: typeof o.plan === 'string' ? o.plan : '', subtasks };