import { Store } from '../store/index.js'; import type { Project, Task, ReviewVerdict } from '../model/types.js'; import { rankByScore } from '../model/scoring.js'; import { createWorktree, worktreeDiff, type WorktreeDiff, type WorktreeInfo } from '../executor/worktree.js'; import { runTask, type RunnerFn } from '../executor/runner.js'; import { runVerify, type VerifyFn } from '../executor/verify.js'; import { reviewCode, reviewSecurity, type ReviewerFn } from '../executor/reviewer.js'; import { pickModel } from '../executor/models.js'; /** 失败后默认最多自动重试次数(项目级 maxRetries 未设置时回退此值) */ export const DEFAULT_MAX_RETRIES = 2; /** 指数退避基础等待(第 n 次重试等待 BASE * 2^(n-1),上限 MAX) */ const BACKOFF_BASE_MS = 30_000; // 30s const BACKOFF_MAX_MS = 10 * 60_000; // 10min /** 第 attempt 次重试(1-indexed)的等待毫秒 */ export function computeBackoffMs(attempt: number): number { return Math.min(BACKOFF_BASE_MS * (2 ** (attempt - 1)), BACKOFF_MAX_MS); } export interface OrchestratorLogger { info(msg: string): void; error(msg: string): void; } /** 依赖注入点:测试传 mock,生产用真实现 */ export interface OrchestratorDeps { runner: RunnerFn; reviewCode: ReviewerFn; // code review(kind=reviewer 的 run) reviewSecurity: ReviewerFn; // 安全审计(kind=security 的 run) verify: VerifyFn; createWorktree: (repoPath: string, taskId: string, baseBranch: string) => Promise; worktreeDiff: (repoPath: string, dir: string, branch: string, baseBranch: string) => Promise; /** 当前时间戳(ms);测试注入可控时钟(默认 Date.now) */ nowMs: () => number; } export interface Orchestrator { /** 跑一轮领取(同步领取 + 异步执行,不阻塞)。错误只记日志。 */ tick(): void; /** 等待所有在途执行收尾(测试用) */ drain(): Promise; /** 在途任务 id(防同任务重复领取) */ readonly inflight: ReadonlyMap; } /** * 编排器核心(与定时器解耦,便于测试)。 * 每轮对 status=active 且 autonomy≠manual 的项目:在途数 < concurrency 时领任务—— * ready 叶子(deps 全 done;auto-easy 只领 easy)或 queued(重试/重启遗留)。 * 成功 → setResult + exec_review;失败 → failed → 重试 ≤MAX_RETRIES 次 → needs_attention。 */ export function createOrchestrator(store: Store, log: OrchestratorLogger, deps: Partial = {}): Orchestrator { const d: OrchestratorDeps = { runner: runTask, reviewCode, reviewSecurity, verify: runVerify, createWorktree, worktreeDiff, nowMs: Date.now, ...deps }; const inflight = new Map(); // taskId → projectId const pending = new Set>(); /** 退避表:taskId → 下次可领取时间戳(ms);daemon 重启后清空,已有 queued 任务即刻可领 */ const backoffUntil = new Map(); /** * 本项目可领取的任务:queued(重试/孤儿)+ ready 叶子且 deps 全 done;auto-easy 只挑 easy。 * 按调度分降序返回(score = 自身分 + 已完成依赖分 + 等待解锁的 blocked 任务分,见 model/scoring.ts)。 */ function claimable(project: Project): Array<{ task: Task; score: number }> { const nowMs = d.nowMs(); const tasks = store.listTasks(project.id); const byId = new Map(tasks.map((t) => [t.id, t])); const parents = new Set(tasks.filter((t) => t.parentId).map((t) => t.parentId as string)); const easyOnly = project.autonomy === 'auto-easy'; const candidates = tasks.filter((t) => { if (inflight.has(t.id)) return false; if (t.status !== 'ready' && t.status !== 'queued') return false; if (parents.has(t.id)) return false; // 非叶子(容器)跳过 if (easyOnly && t.complexity !== 'easy') return false; const until = backoffUntil.get(t.id); if (until !== undefined && nowMs < until) return false; // 退避冷却中 return t.deps.every((dep) => byId.get(dep)?.status === 'done'); }); return rankByScore(candidates, tasks); } /** 跑一个独立复审 run(kind=reviewer/security)。任何失败折叠为 summary=失败原因、verdict=null,不抛错、不挡任务。 */ async function runOneReview( kind: 'reviewer' | 'security', fn: ReviewerFn, task: Task, project: Project, wt: WorktreeInfo, executorReport: string, ): Promise<{ summary: string | null; verdict: ReviewVerdict | null }> { const label = kind === 'reviewer' ? 'code review' : '安全审计'; let reviewRunId: string | null = null; try { const review = store.startRun(task.id, kind, { worktree: wt.dir, branch: wt.branch }); reviewRunId = review.id; log.info(`${label} 任务 ${task.id} run=${review.id} model=${pickModel(task, project, 'reviewer')}`); const rv = await fn(task, project, wt, review.id, executorReport); store.finishRun(review.id, 'succeeded', { transcriptRef: rv.transcriptRef ?? undefined, claudeSessionId: rv.sessionId ?? undefined, }); log.info(`任务 ${task.id} ${label} 完成 verdict=${rv.verdict ?? '(未解析到)'}`); return { summary: rv.summary, verdict: rv.verdict }; } catch (e) { const reMsg = (e as Error).message; if (reviewRunId) { try { store.finishRun(reviewRunId, 'failed', { error: reMsg }); } catch { /* 收尾失败不影响主流程 */ } } log.error(`任务 ${task.id} ${label} 失败(不挡任务,照常进 exec_review):${reMsg}`); return { summary: `自动复审失败:${reMsg}`, verdict: null }; } } /** 单任务全流程:executing → worktree → run → verify → code review run → 安全审计 run(均失败不挡)→ setResult(四字段) → exec_review / failed(重试) */ async function executeTask(project: Project, task: Task): Promise { let runId: string | null = null; let runClosed = false; try { store.transition(task.id, 'executing', { by: 'orchestrator' }); const wt = await d.createWorktree(project.repoPath, task.id, project.defaultBranch); const run = store.startRun(task.id, 'executor', { worktree: wt.dir, branch: wt.branch }); runId = run.id; log.info(`执行任务 ${task.id}「${task.title}」 run=${run.id} worktree=${wt.dir}`); const rr = await d.runner(task, project, wt, run.id); if (!rr.ok) { store.finishRun(run.id, 'failed', { error: rr.error ?? '执行失败', transcriptRef: rr.transcriptRef ?? undefined, claudeSessionId: rr.sessionId ?? undefined, }); runClosed = true; throw new Error(rr.error ?? '执行失败'); } const vr = await d.verify(project, wt.dir, run.id); if (!vr.ok) { store.finishRun(run.id, 'failed', { error: vr.error ?? 'verify 失败', transcriptRef: rr.transcriptRef ?? undefined, claudeSessionId: rr.sessionId ?? undefined, }); runClosed = true; throw new Error(vr.error ?? 'verify 失败'); } const diff = await d.worktreeDiff(project.repoPath, wt.dir, wt.branch, project.defaultBranch); // 双复审(顺序):code review run(kind=reviewer)→ 安全审计 run(kind=security)。任一失败不挡任务。 const code = await runOneReview('reviewer', d.reviewCode, task, project, wt, rr.finalText ?? ''); const sec = await runOneReview('security', d.reviewSecurity, task, project, wt, rr.finalText ?? ''); store.setResult(task.id, { branch: wt.branch, worktree: wt.dir, diffSummary: diff.diffSummary, commits: diff.commits, prUrl: null, summary: code.summary, verdict: code.verdict, securitySummary: sec.summary, securityVerdict: sec.verdict, }); store.transition(task.id, 'exec_review', { by: 'orchestrator', runId: run.id }); store.finishRun(run.id, 'succeeded', { transcriptRef: rr.transcriptRef ?? undefined, claudeSessionId: rr.sessionId ?? undefined, }); log.info(`任务 ${task.id} 执行完成 → exec_review(${diff.commits.length} commits)`); } catch (e) { const msg = (e as Error).message; log.error(`任务 ${task.id} 执行失败:${msg}`); try { if (runId && !runClosed) { store.finishRun(runId, 'failed', { error: msg }); // worktree 创建后抛错(diff 等)时收尾 } else if (!runId) { // run 还没建(如 createWorktree 失败):补记一条 failed run,保证重试计数不漏 const r = store.startRun(task.id, 'executor'); store.finishRun(r.id, 'failed', { error: msg }); } store.transition(task.id, 'failed', { by: 'orchestrator', error: msg }); const allFailed = store.listRuns(task.id).filter((r) => r.kind === 'executor' && r.status === 'failed').length; // 相对于上次手动重投基线的净失败次数(retryBaseline 在 requeueTask 时设置) const netFailed = Math.max(0, allFailed - task.retryBaseline); const priorNetFailed = Math.max(0, netFailed - 1); // 不含本次 const maxRetries = project.maxRetries; if (priorNetFailed < maxRetries) { const attempt = priorNetFailed + 1; // 本次是第几次重试 const backoffMs = computeBackoffMs(attempt); backoffUntil.set(task.id, d.nowMs() + backoffMs); store.transition(task.id, 'queued', { by: 'orchestrator', retry: attempt, backoffMs }); log.info(`任务 ${task.id} 重新入队(第 ${attempt} 次重试,退避 ${backoffMs}ms 后可领)`); } else { backoffUntil.delete(task.id); store.transition(task.id, 'needs_attention', { by: 'orchestrator', allFailed, netFailed }); log.error(`任务 ${task.id} 净失败 ${netFailed} 次(上限 ${maxRetries})→ needs_attention`); } } catch (e2) { log.error(`任务 ${task.id} 失败收尾出错:${(e2 as Error).message}`); } } } function tick(): void { try { for (const p of store.listProjects()) { if (p.status !== 'active' || p.autonomy === 'manual') continue; let active = 0; for (const pid of inflight.values()) if (pid === p.id) active++; if (active >= p.concurrency) continue; for (const { task, score } of claimable(p)) { if (active >= p.concurrency) break; try { if (task.status === 'ready') store.transition(task.id, 'queued', { by: 'orchestrator', score }); } catch (e) { log.error(`任务 ${task.id} 入队失败:${(e as Error).message}`); continue; } log.info(`领取任务 ${task.id}「${task.title}」score=${score} model=${pickModel(task, p, 'executor')}`); inflight.set(task.id, p.id); active++; const job: Promise = executeTask(p, { ...task, status: 'queued' }) .catch((e) => log.error(`任务 ${task.id} 执行异常:${(e as Error).message}`)) .finally(() => { inflight.delete(task.id); pending.delete(job); }); pending.add(job); } } } catch (e) { log.error(`编排器轮询失败:${(e as Error).message}`); } } async function drain(): Promise { while (pending.size > 0) await Promise.allSettled([...pending]); } return { tick, drain, inflight }; } /** * 接线入口:MAESTRO_ORCH_INTERVAL(秒)控制轮询间隔,默认 15,0=关闭。 * 返回 timer 供 shutdown 时 clearInterval。 */ export function startOrchestrator( store: Store, app: { log: OrchestratorLogger }, deps: Partial = {}, ): NodeJS.Timeout | null { const intervalSec = Number(process.env.MAESTRO_ORCH_INTERVAL ?? 15); if (!Number.isFinite(intervalSec) || intervalSec <= 0) { app.log.info('编排器已关闭(MAESTRO_ORCH_INTERVAL=0)'); return null; } const orch = createOrchestrator(store, app.log, deps); const timer = setInterval(() => orch.tick(), intervalSec * 1000); timer.unref(); app.log.info(`编排器已启用:每 ${intervalSec}s 一轮领取(autonomy≠manual 的 active 项目)`); return timer; }