diff --git a/src/executor/pipeline.ts b/src/executor/pipeline.ts index 1e66be5..06c99e3 100644 --- a/src/executor/pipeline.ts +++ b/src/executor/pipeline.ts @@ -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 { + // 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 };