feat: Phase2 完整管线——score 调度 + 自动复审 + 模型分级 + 归档与详情
调度： - model/scoring.ts: score = 自身分(P0=3/P1=2/P2=1) + 已完成依赖分(链条惯性) + 等待解锁的 blocked 任务分(解锁加权)，编排器与 nextExecutable 同一打分 - createTask 校验 deps 存在且同项目(依赖图天然无环) - daemon 重启中断自愈: executing 任务标 failed run 后重新入队(reconcileInterrupted) 执行管线： - executor/cc.ts: 公共 headless CC 执行器(转录/超时/模型回退重试) - executor/reviewer.ts: 执行后自动复审(只读 CC 审 diff)，固定模板 summary (做了什么/怎么做/测试/CodeReview/安全Review/结论) + VERDICT 解析 - executor/models.ts: 按复杂度选模型(easy→sonnet/medium→opus/hard→fable5)， env 可覆盖、project.model 最优先、不可用自动回退链 - runner: 测试/构建命令白名单(npm/go/shellcheck/make/pytest)，prompt 要求实跑测试 - TaskResult 加 summary/verdict; RunKind 加 reviewer - 容器收口: 已拆解 Hard 子任务全 done → 容器自动 done(afterDone 逐级向上) 看板： - 五徽章组(待审批/待执行/执行中/被阻塞/总量，hover 展开，均不含已完成) - 归档区: 深度1整树完成沉底，时间倒序分页(10/20/50/100 chip 选择) - 归档详情对话框: 全属性/执行历史与时长/审批记录/状态流转时间线(含相关人或事) - Agent 面板显示调度模式 + 各复杂度实际模型 - 结果闸展示复审 summary + 建议通过/拒绝徽章 - 筛选修复(组选与单选分离、已拆解移出进行中)、同步按钮收进配置面板、 保存配置自动收起、预览全宽、被依赖阻塞→被阻塞 - API: GET /api/tasks/:id/events(任务级事件时间线)、/api/agents 带 scheduling/models 测试: 49/49(新增 scoring/复审/模型/容器收口/deps 校验/中断恢复) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
wangjia
+6
-1
@@ -4,6 +4,7 @@ import { Store } from '../store/index.js';
|
||||
import { buildServer, attachWebSocket } from '../api/server.js';
|
||||
import { registerStatic } from '../api/static.js';
|
||||
import { loadConfig } from './config.js';
|
||||
import { startOrchestrator } from './orchestrator.js';
|
||||
import { syncProject, todoJsonPath } from '../sync/todo-sync.js';
|
||||
|
||||
/**
|
||||
@@ -46,11 +47,12 @@ function startSyncLoop(store: Store, app: FastifyInstance): NodeJS.Timeout | nul
|
||||
return timer;
|
||||
}
|
||||
|
||||
/** maestrod：核心 daemon。Phase 1 = Store + REST/WS API（手动驱动；编排器在 Phase 2 接入）。 */
|
||||
/** maestrod：核心 daemon。Store + REST/WS API + 定时同步 + 编排器（Phase 2：自动领取可执行任务并在 worktree 起 headless CC）。 */
|
||||
async function main(): Promise<void> {
|
||||
const cfg = loadConfig();
|
||||
const store = new Store(cfg.dbFile);
|
||||
const rec = store.reconcileDeps(); // 启动对账：ready↔blocked 按依赖纠正存量数据
|
||||
const itr = store.reconcileInterrupted(); // 中断恢复：上次退出时在跑的任务重新入队
|
||||
const app = buildServer({ store, logger: true });
|
||||
|
||||
registerStatic(app); // Web 看板（web/ 静态文件）
|
||||
@@ -59,12 +61,15 @@ async function main(): Promise<void> {
|
||||
await app.listen({ host: cfg.host, port: cfg.port });
|
||||
app.log.info(`maestrod 就绪 · db=${cfg.dbFile} · http://${cfg.host}:${cfg.port} · ws ${cfg.host}:${cfg.port}/ws`);
|
||||
if (rec.blocked || rec.released) app.log.info(`依赖对账：转入等依赖 ${rec.blocked} · 放行可执行 ${rec.released}`);
|
||||
if (itr.tasks) app.log.info(`中断恢复：${itr.tasks} 个执行中任务重新入队（${itr.runs} 个 run 标记中断）`);
|
||||
|
||||
const syncTimer = startSyncLoop(store, app);
|
||||
const orchTimer = startOrchestrator(store, app); // 编排器：自动领取可执行任务（MAESTRO_ORCH_INTERVAL 秒，0=关闭）
|
||||
|
||||
const shutdown = async (): Promise<void> => {
|
||||
app.log.info('收到退出信号，关闭中…');
|
||||
if (syncTimer) clearInterval(syncTimer);
|
||||
if (orchTimer) clearInterval(orchTimer);
|
||||
await app.close();
|
||||
store.close();
|
||||
process.exit(0);
|
||||
|
||||
@@ -0,0 +1,224 @@
|
||||
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 { reviewTask, type ReviewerFn } from '../executor/reviewer.js';
|
||||
import { pickModel } from '../executor/models.js';
|
||||
|
||||
/** 失败后最多自动重试次数（重试 2 次 = 最多 3 次执行），之后 → needs_attention */
|
||||
export const MAX_RETRIES = 2;
|
||||
|
||||
export interface OrchestratorLogger {
|
||||
info(msg: string): void;
|
||||
error(msg: string): void;
|
||||
}
|
||||
|
||||
/** 依赖注入点：测试传 mock，生产用真实现 */
|
||||
export interface OrchestratorDeps {
|
||||
runner: RunnerFn;
|
||||
reviewer: ReviewerFn;
|
||||
verify: VerifyFn;
|
||||
createWorktree: (repoPath: string, taskId: string, baseBranch: string) => Promise<WorktreeInfo>;
|
||||
worktreeDiff: (repoPath: string, dir: string, branch: string, baseBranch: string) => Promise<WorktreeDiff>;
|
||||
}
|
||||
|
||||
export interface Orchestrator {
|
||||
/** 跑一轮领取（同步领取 + 异步执行，不阻塞）。错误只记日志。 */
|
||||
tick(): void;
|
||||
/** 等待所有在途执行收尾（测试用） */
|
||||
drain(): Promise<void>;
|
||||
/** 在途任务 id（防同任务重复领取） */
|
||||
readonly inflight: ReadonlyMap<string, string>;
|
||||
}
|
||||
|
||||
/**
|
||||
* 编排器核心（与定时器解耦，便于测试）。
|
||||
* 每轮对 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<OrchestratorDeps> = {}): Orchestrator {
|
||||
const d: OrchestratorDeps = { runner: runTask, reviewer: reviewTask, verify: runVerify, createWorktree, worktreeDiff, ...deps };
|
||||
const inflight = new Map<string, string>(); // taskId → projectId
|
||||
const pending = new Set<Promise<void>>();
|
||||
|
||||
/**
|
||||
* 本项目可领取的任务：queued（重试/孤儿）+ ready 叶子且 deps 全 done；auto-easy 只挑 easy。
|
||||
* 按调度分降序返回（score = 自身分 + 已完成依赖分 + 等待解锁的 blocked 任务分，见 model/scoring.ts）。
|
||||
*/
|
||||
function claimable(project: Project): Array<{ task: Task; score: number }> {
|
||||
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;
|
||||
return t.deps.every((dep) => byId.get(dep)?.status === 'done');
|
||||
});
|
||||
return rankByScore(candidates, tasks);
|
||||
}
|
||||
|
||||
/** 单任务全流程：executing → worktree → run → verify → reviewer(复审，失败不挡) → setResult(summary/verdict) → exec_review / failed(重试) */
|
||||
async function executeTask(project: Project, task: Task): Promise<void> {
|
||||
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);
|
||||
|
||||
// 自动复审（kind=reviewer 的新 run）：失败不挡任务，summary 记失败原因、verdict=null
|
||||
let summary: string | null = null;
|
||||
let verdict: ReviewVerdict | null = null;
|
||||
let reviewRunId: string | null = null;
|
||||
try {
|
||||
const review = store.startRun(task.id, 'reviewer', { worktree: wt.dir, branch: wt.branch });
|
||||
reviewRunId = review.id;
|
||||
log.info(`复审任务 ${task.id} run=${review.id} model=${pickModel(task, project, 'reviewer')}`);
|
||||
const rv = await d.reviewer(task, project, wt, review.id, rr.finalText ?? '');
|
||||
summary = rv.summary;
|
||||
verdict = rv.verdict;
|
||||
store.finishRun(review.id, 'succeeded', {
|
||||
transcriptRef: rv.transcriptRef ?? undefined,
|
||||
claudeSessionId: rv.sessionId ?? undefined,
|
||||
});
|
||||
log.info(`任务 ${task.id} 复审完成 verdict=${verdict ?? '（未解析到）'}`);
|
||||
} catch (e) {
|
||||
const reMsg = (e as Error).message;
|
||||
summary = `自动复审失败：${reMsg}`;
|
||||
verdict = null;
|
||||
if (reviewRunId) {
|
||||
try { store.finishRun(reviewRunId, 'failed', { error: reMsg }); } catch { /* 收尾失败不影响主流程 */ }
|
||||
}
|
||||
log.error(`任务 ${task.id} 复审失败（不挡任务，照常进 exec_review）：${reMsg}`);
|
||||
}
|
||||
|
||||
store.setResult(task.id, {
|
||||
branch: wt.branch, worktree: wt.dir,
|
||||
diffSummary: diff.diffSummary, commits: diff.commits, prUrl: null,
|
||||
summary, 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 failedRuns = store.listRuns(task.id).filter((r) => r.kind === 'executor' && r.status === 'failed').length;
|
||||
const priorFailed = Math.max(0, failedRuns - 1); // 不含本次
|
||||
if (priorFailed < MAX_RETRIES) {
|
||||
store.transition(task.id, 'queued', { by: 'orchestrator', retry: priorFailed + 1 });
|
||||
log.info(`任务 ${task.id} 重新入队（第 ${priorFailed + 1} 次重试，下一轮领取）`);
|
||||
} else {
|
||||
store.transition(task.id, 'needs_attention', { by: 'orchestrator', failedRuns });
|
||||
log.error(`任务 ${task.id} 连续失败 ${failedRuns} 次 → 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<void> = 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<void> {
|
||||
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<OrchestratorDeps> = {},
|
||||
): 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;
|
||||
}
|
||||
Reference in New Issue
Block a user