diff --git a/src/daemon/index.ts b/src/daemon/index.ts index 3648623..4ad33f1 100644 --- a/src/daemon/index.ts +++ b/src/daemon/index.ts @@ -6,6 +6,7 @@ import { registerStatic } from '../api/static.js'; import { loadConfig } from './config.js'; import { startOrchestrator } from './orchestrator.js'; import { startNotifier } from './notify.js'; +import { isWorkerAlive, heartbeatAgeMs } from '../executor/protocol.js'; import { syncProject, todoJsonPath } from '../sync/todo-sync.js'; /** @@ -53,7 +54,15 @@ async function main(): Promise { const cfg = loadConfig(); const store = new Store(cfg.dbFile); const rec = store.reconcileDeps(); // 启动对账:ready↔blocked 按依赖纠正存量数据 - const itr = store.reconcileInterrupted(); // 中断恢复:上次退出时在跑的任务重新入队 + // 中断恢复(多进程执行):用 worker_pid + 心跳真判活——worker 仍活则 re-adopt(daemon 续 ingest 其 outbox), + // 死则回收(failTaskAttempt 收尾 + 重试/needs_attention)。 + const itr = store.reconcileInterrupted((run) => + run !== null && isWorkerAlive({ + pid: run.workerPid, + heartbeatAgeMs: heartbeatAgeMs(run.id), + startedAgeMs: Date.now() - Date.parse(run.startedAt), + }), + ); const app = buildServer({ store, logger: true }); registerStatic(app); // Web 看板(web/ 静态文件) diff --git a/src/daemon/ingest.ts b/src/daemon/ingest.ts new file mode 100644 index 0000000..2829b36 --- /dev/null +++ b/src/daemon/ingest.ts @@ -0,0 +1,114 @@ +import { Store } from '../store/index.js'; +import { readOutboxSince, type OutboxRecord } from '../executor/protocol.js'; + +/** + * Outbox → DB 摄取(daemon 是唯一 DB 写者)。 + * + * worker 全程不碰 DB,只把进度/结果追加进 runs//outbox.ndjson。daemon 在每轮 tick 里把 + * 这些记录读出来,按 seq 升序逐条映射成 DB 写(finishRun / setResult / transition / failTaskAttempt …)。 + * + * 幂等靠 run.lastSeq 游标:只处理 seq>lastSeq 的记录,每处理一条就 setLastSeq(seq)。崩溃重启续读时 + * 已落库的 seq 不会重做——故 result/failed/done 这类终态记录即便文件还在,也只会被消费一次。 + */ +export interface IngestLogger { + info(msg: string): void; + error(msg: string): void; +} + +/** + * 摄取单个 run 的新 outbox 记录(seq>run.lastSeq),按序落库。 + * run 不存在则忽略(已被删/异常)。每条处理后推进 lastSeq 游标。 + */ +export function ingestRun(store: Store, log: IngestLogger, runId: string): void { + const run = store.getRun(runId); + if (!run) return; + const taskId = run.taskId; + const recs = readOutboxSince(runId, run.lastSeq); + for (const rec of recs) { + try { + applyRecord(store, log, taskId, runId, rec); + } catch (e) { + // 单条映射失败不阻断后续 run 的摄取;但本条不推进游标,下轮重试。 + log.error(`ingest run=${runId} seq=${rec.seq} 失败:${(e as Error).message}`); + return; + } + store.setLastSeq(runId, rec.seq); + } +} + +/** 把一条 OutboxRecord 映射为 DB 写。 */ +function applyRecord(store: Store, log: IngestLogger, taskId: string, runId: string, rec: OutboxRecord): void { + switch (rec.type) { + case 'started': + // worker 启动自报;pid 已由 daemon spawn 时 setWorkerPid 写过,这里仅记日志。 + log.info(`worker 启动 task=${taskId} run=${runId} pid=${rec.pid} model=${rec.model}`); + return; + + case 'phase': + // 进度阶段,仅日志(看板事件可后续接)。 + log.info(`task=${taskId} run=${runId} 阶段=${rec.phase}`); + return; + + case 'failed': { + // 先把 executor run 收尾为 failed(带转录/会话),再走失败/重试策略。 + // failTaskAttempt 见 run 已 ended(非 started)不会重复收尾,只做重试决策。 + try { + store.finishRun(runId, 'failed', { + error: rec.error, + transcriptRef: rec.transcriptRef ?? undefined, + claudeSessionId: rec.sessionId ?? undefined, + }); + } catch (e) { + log.error(`task=${taskId} run=${runId} 收尾 failed 出错(继续重试决策):${(e as Error).message}`); + } + store.failTaskAttempt(taskId, runId, rec.error); + log.info(`task=${taskId} run=${runId} 执行失败:${rec.error}`); + return; + } + + case 'result': { + // 成功终态:建 reviewer/security 两条复审 run + 收尾,写 result 四字段,转 exec_review,收尾 executor run。 + const cr = store.startRun(taskId, 'reviewer', { worktree: rec.worktree, branch: rec.branch }); + store.finishRun(cr.id, 'succeeded', { transcriptRef: rec.code.transcriptRef ?? undefined }); + + const sr = store.startRun(taskId, 'security', { worktree: rec.worktree, branch: rec.branch }); + store.finishRun(sr.id, 'succeeded', { transcriptRef: rec.security.transcriptRef ?? undefined }); + + store.setResult(taskId, { + branch: rec.branch, + worktree: rec.worktree, + diffSummary: rec.diffSummary, + commits: rec.commits, + prUrl: null, + summary: rec.code.summary, + verdict: rec.code.verdict, + securitySummary: rec.security.summary, + securityVerdict: rec.security.verdict, + mergeTaskId: null, + }); + store.transition(taskId, 'exec_review', { by: 'ingest', runId }); + store.finishRun(runId, 'succeeded', { + transcriptRef: rec.executor.transcriptRef ?? undefined, + claudeSessionId: rec.executor.sessionId ?? undefined, + }); + log.info(`task=${taskId} run=${runId} 执行完成 → exec_review(${rec.commits.length} commits)`); + return; + } + + case 'done': + // worker 即将退出:终态标记。run 已由 result/failed 收尾,这里无须额外 DB 写。 + log.info(`task=${taskId} run=${runId} worker 退出`); + return; + } +} + +/** + * 摄取所有在途 run:对每个 executing 任务的最近一条 executor run 调 ingestRun。 + * daemon 每轮 tick 调一次,把 worker 期间累积的 outbox 落库。 + */ +export function ingestAll(store: Store, log: IngestLogger): void { + for (const { run } of store.executingWithLatestExecutorRun()) { + if (!run) continue; + ingestRun(store, log, run.id); + } +} diff --git a/src/daemon/orchestrator.ts b/src/daemon/orchestrator.ts index ea4ca8a..3a7edcb 100644 --- a/src/daemon/orchestrator.ts +++ b/src/daemon/orchestrator.ts @@ -1,66 +1,107 @@ +import { spawn } from 'node:child_process'; +import { fileURLToPath } from 'node:url'; +import { dirname, join } from 'node:path'; import { Store } from '../store/index.js'; -import type { Project, Task, ReviewVerdict } from '../model/types.js'; +import type { Project, Task } 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 { branchFor, worktreeDirFor } from '../executor/worktree.js'; +import { writeJobSpec, isWorkerAlive, heartbeatAgeMs, type JobSpec } from '../executor/protocol.js'; import { pickModel } from '../executor/models.js'; +import { ingestAll, type IngestLogger } from './ingest.js'; -/** 失败后默认最多自动重试次数(项目级 maxRetries 未设置时回退此值) */ +/** 失败后默认最多自动重试次数(项目级 maxRetries 未设置时回退此值)。失败/退避策略本体在 store.failTaskAttempt。 */ 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,生产用真实现 */ +/** + * worker 入口解析:返回 spawn 用的 [cmd, ...args](不含 runId,由调用方追加)。 + * - 默认:node <本文件同级 ../executor/worker.js>(编译后 dist 布局:daemon/ 与 executor/ 同级)。 + * - env MAESTRO_WORKER_CMD 覆盖整条命令(空格分隔),供测试 / tsx 跑 .ts 入口用, + * 例:MAESTRO_WORKER_CMD="npx tsx src/executor/worker.ts"。 + */ +export function workerEntry(): string[] { + const override = process.env.MAESTRO_WORKER_CMD?.trim(); + if (override) return override.split(/\s+/); + const here = dirname(fileURLToPath(import.meta.url)); + return ['node', join(here, '..', 'executor', 'worker.js')]; +} + +/** + * spawn worker 时传给子进程的最小 env:只保留 PATH/HOME/LANG 与 ANTHROPIC_ / CLAUDE_ 前缀变量, + * 不把 daemon 全量 env(含 MAESTRO_ 控制变量、无关密钥)漏给独立 worker 进程。 + * MAESTRO_DATA_DIR 例外透传——worker 必须和 daemon 看同一个 runs/ 目录(job.json/outbox 路径一致)。 + */ +export function workerEnv(src: NodeJS.ProcessEnv = process.env): NodeJS.ProcessEnv { + const out: NodeJS.ProcessEnv = {}; + for (const k of ['PATH', 'HOME', 'LANG', 'MAESTRO_DATA_DIR', 'MAESTRO_WORKER_CMD'] as const) { + if (src[k] !== undefined) out[k] = src[k]; + } + for (const [k, v] of Object.entries(src)) { + if (v !== undefined && (k.startsWith('ANTHROPIC_') || k.startsWith('CLAUDE_'))) out[k] = v; + } + return out; +} + +/** 默认 spawnWorker:detached + 脱离 stdio + 收敛 env,unref 后自存活(daemon 退出不杀 worker)。返回子进程 pid。 */ +function defaultSpawnWorker(runId: string): number { + const [cmd, ...args] = workerEntry(); + const child = spawn(cmd, [...args, runId], { + detached: true, + stdio: 'ignore', + env: workerEnv(), + }); + child.unref(); + if (child.pid === undefined) throw new Error('spawn worker 未返回 pid'); + return child.pid; +} + +/** 依赖注入点:生产用默认实现,测试传 mock(不真 spawn / 不真判活 / ingest no-op)。 */ 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) */ + /** spawn 一个独立 worker 进程跑该 run,返回其 pid。 */ + spawnWorker: (runId: string) => number; + /** worker 是否存活(reaper 判死用),默认 protocol.isWorkerAlive。 */ + isWorkerAlive: typeof isWorkerAlive; + /** 把所有在途 run 的 outbox 摄取入库(daemon 唯一写者),默认 ingest.ingestAll。 */ + ingestAll: (store: Store, log: IngestLogger) => void; + /** 写 worker 输入 job.json,默认 protocol.writeJobSpec(测试可 mock,避免落盘)。 */ + writeJobSpec: (job: JobSpec) => void; + /** 当前时间戳(ms);测试注入可控时钟(默认 Date.now)。 */ nowMs: () => number; } export interface Orchestrator { - /** 跑一轮领取(同步领取 + 异步执行,不阻塞)。错误只记日志。 */ + /** 跑一轮:先 ingest(outbox→DB)+ reaper(回收死 worker),再领新任务 spawn worker。错误只记日志、不抛。 */ tick(): void; - /** 等待所有在途执行收尾(测试用) */ - drain(): Promise; - /** 在途任务 id(防同任务重复领取) */ - readonly inflight: ReadonlyMap; + /** 仅领取/spawn 那一步(测试细粒度断言用,不含 ingest/reaper)。 */ + claimTick(): void; + /** ingest 所有在途 run(测试用)。 */ + ingest(): void; + /** 回收死 worker(测试用)。 */ + reap(): void; } /** - * 编排器核心(与定时器解耦,便于测试)。 - * 每轮对 status=active 且 autonomy≠manual 的项目:在途数 < concurrency 时领任务—— - * ready 叶子(deps 全 done;auto-easy 只领 easy)或 queued(重试/重启遗留)。 - * 成功 → setResult + exec_review;失败 → failed → 重试 ≤MAX_RETRIES 次 → needs_attention。 + * 编排器(监工):自身不跑 agent,只领任务 + spawn 独立 worker + 把 worker 的 outbox 摄取入库 + 回收死 worker。 + * 全部 DB 写都在 daemon 进程内(含 ingest 落的 run/result/transition),WS 推送照常经 store.subscribe 触发。 + * 无内存在途态——并发闸读 store.countExecuting,退避读 task.nextEligibleAt,故跨 daemon 重启天然持久。 */ 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(); + const d: OrchestratorDeps = { + spawnWorker: defaultSpawnWorker, + isWorkerAlive, + ingestAll, + writeJobSpec, + nowMs: Date.now, + ...deps, + }; /** * 本项目可领取的任务:queued(重试/孤儿)+ ready 叶子且 deps 全 done;auto-easy 只挑 easy。 - * 按调度分降序返回(score = 自身分 + 已完成依赖分 + 等待解锁的 blocked 任务分,见 model/scoring.ts)。 + * 退避读持久化的 task.nextEligibleAt(早于它不领)。按调度分降序返回(见 model/scoring.ts)。 */ function claimable(project: Project): Array<{ task: Task; score: number }> { const nowMs = d.nowMs(); @@ -69,178 +110,113 @@ export function createOrchestrator(store: Store, log: OrchestratorLogger, deps: 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; // 退避冷却中 + if (t.nextEligibleAt && nowMs < Date.parse(t.nextEligibleAt)) 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 { + /** 领取一个任务:建分支/worktree 路径 → executing → startRun → 写 job.json → spawn worker → 记 pid。 */ + function claimOne(project: Project, task: Task, score: number): void { let runId: string | null = null; - let runClosed = false; try { + if (task.status === 'ready') store.transition(task.id, 'queued', { by: 'orchestrator', score }); + + const branch = branchFor(task.id); + const dir = worktreeDirFor(project.repoPath, task.id); + 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 }); + const run = store.startRun(task.id, 'executor', { worktree: dir, 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, - mergeTaskId: null, + d.writeJobSpec({ + runId: run.id, + task: { ...task, status: 'executing' }, + project, + worktreeDir: dir, + branch, }); - 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)`); + + log.info(`领取任务 ${task.id}「${task.title}」score=${score} run=${run.id} model=${pickModel(task, project, 'executor')}`); + const pid = d.spawnWorker(run.id); + store.setWorkerPid(run.id, pid); + log.info(`任务 ${task.id} 已起 worker pid=${pid} worktree=${dir}`); } catch (e) { const msg = (e as Error).message; - log.error(`任务 ${task.id} 执行失败:${msg}`); + log.error(`任务 ${task.id} 领取/spawn 失败:${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`); - } + store.failTaskAttempt(task.id, runId, msg); } catch (e2) { log.error(`任务 ${task.id} 失败收尾出错:${(e2 as Error).message}`); } } } - function tick(): void { + /** 领取轮:对每个 active 且 autonomy≠manual 的项目,在 countExecuting= 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); + claimOne(p, task, score); 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}`); + log.error(`编排器领取轮失败:${(e as Error).message}`); } } - async function drain(): Promise { - while (pending.size > 0) await Promise.allSettled([...pending]); + /** ingest 所有在途 run 的 outbox(daemon 唯一 DB 写者把 worker 进度/结果落库)。 */ + function ingest(): void { + try { + d.ingestAll(store, log); + } catch (e) { + log.error(`ingest 轮失败:${(e as Error).message}`); + } } - return { tick, drain, inflight }; + /** 回收:对每个 executing 任务判活,死 worker → failTaskAttempt(收尾 + 重试/needs_attention)。 */ + function reap(): void { + try { + for (const { task, run } of store.executingWithLatestExecutorRun()) { + const alive = run !== null && d.isWorkerAlive({ + pid: run.workerPid, + heartbeatAgeMs: heartbeatAgeMs(run.id, d.nowMs()), + startedAgeMs: d.nowMs() - Date.parse(run.startedAt), + }); + if (alive) continue; + log.error(`任务 ${task.id} worker 异常退出(run=${run?.id ?? '无'})→ 回收`); + try { + store.failTaskAttempt(task.id, run?.id ?? null, 'worker 异常退出'); + } catch (e) { + log.error(`任务 ${task.id} 回收收尾出错:${(e as Error).message}`); + } + } + } catch (e) { + log.error(`编排器回收轮失败:${(e as Error).message}`); + } + } + + function tick(): void { + ingest(); // 先把 worker 进度/结果落库(可能把 executing→exec_review,腾出并发槽) + reap(); // 再回收死 worker(可能把 executing→queued/needs_attention,腾出并发槽) + claimTick(); // 最后领新任务 + } + + return { tick, claimTick, ingest, reap }; } /** * 接线入口:MAESTRO_ORCH_INTERVAL(秒)控制轮询间隔,默认 15,0=关闭。 - * 返回 timer 供 shutdown 时 clearInterval。 + * 每轮依次 ingest → reaper → 领取(见 tick)。返回 timer 供 shutdown 时 clearInterval。 */ export function startOrchestrator( store: Store, @@ -255,6 +231,6 @@ export function startOrchestrator( const orch = createOrchestrator(store, app.log, deps); const timer = setInterval(() => orch.tick(), intervalSec * 1000); timer.unref(); - app.log.info(`编排器已启用:每 ${intervalSec}s 一轮领取(autonomy≠manual 的 active 项目)`); + app.log.info(`编排器已启用:每 ${intervalSec}s 一轮(ingest→回收→领取,autonomy≠manual 的 active 项目)`); return timer; } diff --git a/test/ingest.test.ts b/test/ingest.test.ts new file mode 100644 index 0000000..043851c --- /dev/null +++ b/test/ingest.test.ts @@ -0,0 +1,205 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { Store } from '../src/store/index.js'; +import { ingestRun } from '../src/daemon/ingest.js'; +import { appendOutbox, type OutboxPayload } from '../src/executor/protocol.js'; +import { branchFor, worktreeDirFor } from '../src/executor/worktree.js'; + +const noopLog = { info: (): void => undefined, error: (): void => undefined }; + +/** + * 隔离的 MAESTRO_DATA_DIR(outbox 落在临时目录),跑回调,结束清理。 + * protocol.ts 的 runDir/outboxPath 在调用时读 env,故进 cb 前设好、cb 内的 append/ingest 都命中临时目录。 + */ +function withTmpDataDir(cb: () => void): void { + const dir = mkdtempSync(join(tmpdir(), 'maestro-ingest-')); + const prev = process.env.MAESTRO_DATA_DIR; + process.env.MAESTRO_DATA_DIR = dir; + try { + cb(); + } finally { + if (prev === undefined) delete process.env.MAESTRO_DATA_DIR; + else process.env.MAESTRO_DATA_DIR = prev; + rmSync(dir, { recursive: true, force: true }); + } +} + +/** 真 Store + 项目 + easy 任务推进到 executing,并建一条 executor run(模拟 daemon 已领取、worker 已 spawn)。 */ +function setupExecuting(opts: { autonomy?: 'auto-easy' | 'auto-approved'; maxRetries?: number } = {}): { + store: Store; projectId: string; taskId: string; runId: string; dir: string; branch: string; +} { + const store = new Store(':memory:'); + const p = store.createProject({ + name: 'ingest', repoPath: '/tmp/ingest-repo-' + Math.random(), + autonomy: opts.autonomy ?? 'auto-easy', maxRetries: opts.maxRetries, + }); + const t = store.createTask({ projectId: p.id, title: 'tweak', complexity: 'easy' }); + store.setOperations(t.id, '在 README.md 追加一行'); + const branch = branchFor(t.id); + const dir = worktreeDirFor(p.repoPath, t.id); + store.transition(t.id, 'queued', { by: 'test' }); + store.transition(t.id, 'executing', { by: 'test' }); + const run = store.startRun(t.id, 'executor', { worktree: dir, branch }); + store.setWorkerPid(run.id, 4242); + return { store, projectId: p.id, taskId: t.id, runId: run.id, dir, branch }; +} + +function resultPayload(branch: string, worktree: string): OutboxPayload { + return { + type: 'result', + branch, worktree, + diffSummary: ' README.md | 1 +', + commits: ['abc1234 hello maestro'], + executor: { transcriptRef: '/tmp/exec.jsonl', sessionId: 'sess-exec-1' }, + code: { summary: '## 做了什么\nmock 复审通过', verdict: 'approve', transcriptRef: '/tmp/code.jsonl' }, + security: { summary: '## 安全审计\nmock 审计通过', verdict: 'approve', transcriptRef: '/tmp/sec.jsonl' }, + }; +} + +test('ingest result:task→exec_review,setResult 四字段正确,reviewer/security/executor 各 1 条 succeeded', () => { + withTmpDataDir(() => { + const { store, taskId, runId, dir, branch } = setupExecuting(); + + appendOutbox(runId, { type: 'started', pid: 4242, worktree: dir, branch, model: 'claude-sonnet-4-6' }); + appendOutbox(runId, { type: 'phase', phase: 'executing' }); + appendOutbox(runId, resultPayload(branch, dir)); + appendOutbox(runId, { type: 'done' }); + + ingestRun(store, noopLog, runId); + + const done = store.getTask(taskId)!; + assert.equal(done.status, 'exec_review'); + assert.deepEqual(done.result, { + branch, + worktree: dir, + diffSummary: ' README.md | 1 +', + commits: ['abc1234 hello maestro'], + prUrl: null, + summary: '## 做了什么\nmock 复审通过', + verdict: 'approve', + securitySummary: '## 安全审计\nmock 审计通过', + securityVerdict: 'approve', + mergeTaskId: null, + }); + + const runs = store.listRuns(taskId); + const executor = runs.find((r) => r.kind === 'executor')!; + assert.equal(executor.status, 'succeeded'); + assert.equal(executor.transcriptRef, '/tmp/exec.jsonl'); + assert.equal(executor.claudeSessionId, 'sess-exec-1'); + const reviewer = runs.filter((r) => r.kind === 'reviewer'); + assert.equal(reviewer.length, 1); + assert.equal(reviewer[0].status, 'succeeded'); + assert.equal(reviewer[0].transcriptRef, '/tmp/code.jsonl'); + const security = runs.filter((r) => r.kind === 'security'); + assert.equal(security.length, 1); + assert.equal(security[0].status, 'succeeded'); + assert.equal(security[0].transcriptRef, '/tmp/sec.jsonl'); + + // 游标推进到末条(done.seq=4) + assert.equal(store.getRun(runId)!.lastSeq, 4); + store.close(); + }); +}); + +test('ingest result:任一 verdict=reject 也照常落进 result(裁决归用户)', () => { + withTmpDataDir(() => { + const { store, taskId, runId, dir, branch } = setupExecuting(); + appendOutbox(runId, { + ...resultPayload(branch, dir), + code: { summary: '发现问题', verdict: 'reject', transcriptRef: '/tmp/code.jsonl' }, + security: { summary: '发现密钥泄露', verdict: 'reject', transcriptRef: '/tmp/sec.jsonl' }, + } as OutboxPayload); + ingestRun(store, noopLog, runId); + + const done = store.getTask(taskId)!; + assert.equal(done.status, 'exec_review'); + assert.equal(done.result!.verdict, 'reject'); + assert.equal(done.result!.summary, '发现问题'); + assert.equal(done.result!.securityVerdict, 'reject'); + assert.equal(done.result!.securitySummary, '发现密钥泄露'); + store.close(); + }); +}); + +test('ingest failed:首次失败 → task 转 queued + 持久化退避(nextEligibleAt 有值)+ executor run failed', () => { + withTmpDataDir(() => { + const { store, taskId, runId } = setupExecuting(); + + appendOutbox(runId, { type: 'started', pid: 4242, worktree: '/x', branch: 'b', model: 'm' }); + appendOutbox(runId, { type: 'failed', error: 'boom #1', transcriptRef: '/tmp/f.jsonl', sessionId: 'sess-f-1' }); + appendOutbox(runId, { type: 'done' }); + + ingestRun(store, noopLog, runId); + + const t = store.getTask(taskId)!; + assert.equal(t.status, 'queued'); // 第一次失败 → 重入队(默认 maxRetries=2) + assert.ok(t.nextEligibleAt, 'nextEligibleAt 应被持久化(退避)'); + assert.ok(Date.parse(t.nextEligibleAt!) > Date.now() - 1000, 'nextEligibleAt 在未来'); + + const executor = store.listRuns(taskId).find((r) => r.kind === 'executor')!; + assert.equal(executor.status, 'failed'); + assert.match(executor.error ?? '', /boom #1/); + assert.equal(executor.transcriptRef, '/tmp/f.jsonl'); + assert.equal(executor.claudeSessionId, 'sess-f-1'); + store.close(); + }); +}); + +test('ingest failed:maxRetries=0 → 首次失败直接 needs_attention', () => { + withTmpDataDir(() => { + const { store, taskId, runId } = setupExecuting({ maxRetries: 0 }); + appendOutbox(runId, { type: 'failed', error: 'boom', transcriptRef: null, sessionId: null }); + ingestRun(store, noopLog, runId); + assert.equal(store.getTask(taskId)!.status, 'needs_attention'); + store.close(); + }); +}); + +test('幂等:同一批 outbox ingest 两次,DB 不重复变更(lastSeq 生效)', () => { + withTmpDataDir(() => { + const { store, taskId, runId, dir, branch } = setupExecuting(); + appendOutbox(runId, { type: 'started', pid: 4242, worktree: dir, branch, model: 'm' }); + appendOutbox(runId, resultPayload(branch, dir)); + appendOutbox(runId, { type: 'done' }); + + ingestRun(store, noopLog, runId); + const after1 = store.getTask(taskId)!; + const runs1 = store.listRuns(taskId); + assert.equal(after1.status, 'exec_review'); + assert.equal(runs1.filter((r) => r.kind === 'reviewer').length, 1); + assert.equal(runs1.filter((r) => r.kind === 'security').length, 1); + const lastSeq1 = store.getRun(runId)!.lastSeq; + + // 第二次 ingest:seq 全部 ≤ lastSeq → 不处理任何记录 + ingestRun(store, noopLog, runId); + const runs2 = store.listRuns(taskId); + assert.equal(runs2.length, runs1.length, '复审 run 不应重复创建'); + assert.equal(runs2.filter((r) => r.kind === 'reviewer').length, 1); + assert.equal(runs2.filter((r) => r.kind === 'security').length, 1); + assert.equal(store.getRun(runId)!.lastSeq, lastSeq1, 'lastSeq 不变'); + assert.equal(store.getTask(taskId)!.status, 'exec_review'); + store.close(); + }); +}); + +test('ingest 续读:先 ingest 半截(started/phase),再追加 result,第二次 ingest 完成 exec_review', () => { + withTmpDataDir(() => { + const { store, taskId, runId, dir, branch } = setupExecuting(); + appendOutbox(runId, { type: 'started', pid: 4242, worktree: dir, branch, model: 'm' }); + appendOutbox(runId, { type: 'phase', phase: 'executing' }); + ingestRun(store, noopLog, runId); + assert.equal(store.getTask(taskId)!.status, 'executing'); // 还没到 result,仍 executing + assert.equal(store.getRun(runId)!.lastSeq, 2); + + appendOutbox(runId, resultPayload(branch, dir)); + appendOutbox(runId, { type: 'done' }); + ingestRun(store, noopLog, runId); + assert.equal(store.getTask(taskId)!.status, 'exec_review'); + assert.equal(store.getRun(runId)!.lastSeq, 4); + store.close(); + }); +}); diff --git a/test/orchestrator.test.ts b/test/orchestrator.test.ts index c0a0c81..0320332 100644 --- a/test/orchestrator.test.ts +++ b/test/orchestrator.test.ts @@ -1,53 +1,44 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; import { Store } from '../src/store/index.js'; -import { createOrchestrator, DEFAULT_MAX_RETRIES, computeBackoffMs, type OrchestratorDeps } from '../src/daemon/orchestrator.js'; -import type { RunnerResult } from '../src/executor/runner.js'; -import type { ReviewResult } from '../src/executor/reviewer.js'; +import { + createOrchestrator, workerEntry, workerEnv, DEFAULT_MAX_RETRIES, + type OrchestratorDeps, +} from '../src/daemon/orchestrator.js'; +import type { JobSpec } from '../src/executor/protocol.js'; import type { Autonomy } from '../src/model/types.js'; const noopLog = { info: (): void => undefined, error: (): void => undefined }; -function deferred(): { promise: Promise; resolve: (v: T) => void } { - let resolve!: (v: T) => void; - const promise = new Promise((r) => { resolve = r; }); - return { promise, resolve }; +/** 可前进的测试时钟。 */ +function makeClock(initialMs = Date.now()): { nowMs: () => number; advance: (ms: number) => void } { + let t = initialMs; + return { nowMs: () => t, advance: (ms) => { t += ms; } }; } -const okRun: RunnerResult = { ok: true, transcriptRef: '/tmp/fake.jsonl', sessionId: 'sess-mock-1', finalText: '执行自述:改了 README' }; +interface MockState { + spawned: string[]; // 被 spawn 的 runId 列表 + jobs: JobSpec[]; // 被 writeJobSpec 的 job 列表 + ingestCalls: number; + alive: boolean; // isWorkerAlive 的返回(reaper 用) + nextPid: number; +} -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 = {}): OrchestratorDeps { - 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 }), - runner: async () => okRun, - reviewCode: async () => okReview, - reviewSecurity: async () => okSecurity, +/** 全 mock 依赖:不真 spawn / 不真判活 / ingest no-op / job 不落盘。可按用例覆盖。 */ +function mockDeps(state: MockState, overrides: Partial = {}): { deps: OrchestratorDeps } { + const deps: OrchestratorDeps = { + spawnWorker: (runId: string): number => { state.spawned.push(runId); return state.nextPid++; }, + isWorkerAlive: () => state.alive, + ingestAll: () => { state.ingestCalls++; }, + writeJobSpec: (job: JobSpec) => { state.jobs.push(job); }, nowMs: Date.now, ...overrides, }; + return { deps }; } -/** 可前进的测试时钟(用于退避相关测试) */ -function makeClock(initialMs = 0): { nowMs: () => number; advance: (ms: number) => void } { - let t = initialMs; - return { nowMs: () => t, advance: (ms) => { t += ms; } }; +function freshState(): MockState { + return { spawned: [], jobs: [], ingestCalls: 0, alive: true, nextPid: 1000 }; } function setup(autonomy: Autonomy, concurrency = 1): { store: Store; projectId: string } { @@ -58,25 +49,139 @@ function setup(autonomy: Autonomy, concurrency = 1): { store: Store; projectId: return { store, projectId: p.id }; } -async function settle(): Promise { - await new Promise((r) => setImmediate(r)); -} +// ───────────────────────── workerEntry / workerEnv(可读测的小函数)───────────────────────── -test('autonomy=manual:编排器不领取任何任务', async () => { - const { store, projectId } = setup('manual'); - const t = store.createTask({ projectId, title: 'easy task', complexity: 'easy' }); - let calls = 0; - const orch = createOrchestrator(store, noopLog, mockDeps({ - runner: async () => { calls++; return okRun; }, - })); +test('workerEntry:默认指向 ../executor/worker.js(node 运行)', () => { + const prev = process.env.MAESTRO_WORKER_CMD; + delete process.env.MAESTRO_WORKER_CMD; + const [cmd, scriptPath] = workerEntry(); + assert.equal(cmd, 'node'); + assert.match(scriptPath, /executor[/\\]worker\.js$/); + if (prev !== undefined) process.env.MAESTRO_WORKER_CMD = prev; +}); + +test('workerEntry:MAESTRO_WORKER_CMD 覆盖整条命令(空格分隔)', () => { + const prev = process.env.MAESTRO_WORKER_CMD; + process.env.MAESTRO_WORKER_CMD = 'npx tsx src/executor/worker.ts'; + assert.deepEqual(workerEntry(), ['npx', 'tsx', 'src/executor/worker.ts']); + if (prev === undefined) delete process.env.MAESTRO_WORKER_CMD; + else process.env.MAESTRO_WORKER_CMD = prev; +}); + +test('workerEnv:只透传 PATH/HOME/LANG + ANTHROPIC_/CLAUDE_ 前缀;丢弃无关与 MAESTRO_ 控制变量', () => { + const env = workerEnv({ + PATH: '/usr/bin', HOME: '/home/u', LANG: 'en_US.UTF-8', + ANTHROPIC_API_KEY: 'sk-x', CLAUDE_CODE_FOO: 'y', + MAESTRO_ORCH_INTERVAL: '15', MAESTRO_DATA_DIR: '/data', + SOME_SECRET: 'leak', AWS_SECRET_ACCESS_KEY: 'nope', + }); + assert.equal(env.PATH, '/usr/bin'); + assert.equal(env.HOME, '/home/u'); + assert.equal(env.LANG, 'en_US.UTF-8'); + assert.equal(env.ANTHROPIC_API_KEY, 'sk-x'); + assert.equal(env.CLAUDE_CODE_FOO, 'y'); + assert.equal(env.MAESTRO_DATA_DIR, '/data'); // 例外:worker 须看同一 runs/ 目录 + assert.equal(env.MAESTRO_ORCH_INTERVAL, undefined, 'daemon 控制变量不下传'); + assert.equal(env.SOME_SECRET, undefined, '无关变量不下传'); + assert.equal(env.AWS_SECRET_ACCESS_KEY, undefined, '无关密钥不下传'); +}); + +// ───────────────────────── tick 顺序:ingest + reaper 先于领取 ───────────────────────── + +test('tick:每轮先 ingest 再 reaper 再领取', () => { + const { store, projectId } = setup('auto-easy'); + store.createTask({ projectId, title: 'a', complexity: 'easy' }); + const state = freshState(); + const { deps } = mockDeps(state); + const orch = createOrchestrator(store, noopLog, deps); orch.tick(); - await orch.drain(); - assert.equal(calls, 0); + assert.equal(state.ingestCalls, 1, '每轮 ingest 一次'); + assert.equal(state.spawned.length, 1, '领取并 spawn 一个'); +}); + +// ───────────────────────── 领取(监工)───────────────────────── + +test('autonomy=manual:不领取', () => { + const { store, projectId } = setup('manual'); + const t = store.createTask({ projectId, title: 'easy', complexity: 'easy' }); + const state = freshState(); + const { deps } = mockDeps(state); + createOrchestrator(store, noopLog, deps).tick(); + assert.equal(state.spawned.length, 0); assert.equal(store.getTask(t.id)!.status, 'ready'); store.close(); }); -test('autonomy=auto-easy:只领 easy,medium ready 不动', async () => { +test('project paused:不领取', () => { + const { store, projectId } = setup('auto-approved'); + const t = store.createTask({ projectId, title: 'x', complexity: 'easy' }); + store.patchProject(projectId, { status: 'paused' }); + const state = freshState(); + const { deps } = mockDeps(state); + createOrchestrator(store, noopLog, deps).tick(); + assert.equal(state.spawned.length, 0); + assert.equal(store.getTask(t.id)!.status, 'ready'); + store.close(); +}); + +test('并发闸 concurrency=1:一轮只领一个;领取后 executing + 建 executor run + setWorkerPid + writeJobSpec', () => { + const { store, projectId } = setup('auto-approved', 1); + const t1 = store.createTask({ projectId, title: 'a', complexity: 'easy' }); + const t2 = store.createTask({ projectId, title: 'b', complexity: 'easy' }); + const state = freshState(); + const { deps } = mockDeps(state); + const orch = createOrchestrator(store, noopLog, deps); + + orch.claimTick(); + // 并发=1:只领一个 + assert.equal(state.spawned.length, 1); + assert.equal(state.jobs.length, 1); + + // 被领的那个进 executing,另一个仍 ready + const a = store.getTask(t1.id)!; + const b = store.getTask(t2.id)!; + const executing = a.status === 'executing' ? a : b; + const stillReady = a.status === 'executing' ? b : a; + assert.equal(executing.status, 'executing'); + assert.equal(stillReady.status, 'ready'); + + // 建了 executor run + setWorkerPid + writeJobSpec 内容正确 + const runs = store.listRuns(executing.id); + assert.equal(runs.length, 1); + const run = runs[0]; + assert.equal(run.kind, 'executor'); + assert.equal(run.status, 'started'); + assert.equal(run.branch, `maestro/${executing.id}`); + assert.ok(run.workerPid, 'setWorkerPid 已写 pid'); + assert.equal(state.spawned[0], run.id, 'spawn 的 runId == 新建 executor run'); + + const job = state.jobs[0]; + assert.equal(job.runId, run.id); + assert.equal(job.task.id, executing.id); + assert.equal(job.task.status, 'executing'); + assert.equal(job.branch, `maestro/${executing.id}`); + assert.equal(job.worktreeDir, run.worktree); + + // 槽位占满 → 下一轮不再领(仍只 spawn 过一个) + orch.claimTick(); + assert.equal(state.spawned.length, 1, '并发=1,executing 占满 → 不再领'); + store.close(); +}); + +test('并发 concurrency=2:一轮领满两个', () => { + const { store, projectId } = setup('auto-approved', 2); + store.createTask({ projectId, title: 'a', complexity: 'easy' }); + store.createTask({ projectId, title: 'b', complexity: 'easy' }); + store.createTask({ projectId, title: 'c', complexity: 'easy' }); + const state = freshState(); + const { deps } = mockDeps(state); + createOrchestrator(store, noopLog, deps).claimTick(); + assert.equal(state.spawned.length, 2, '并发=2 一轮领两个'); + assert.equal(store.listTasks(projectId).filter((t) => t.status === 'executing').length, 2); + store.close(); +}); + +test('auto-easy:只领 easy,medium ready 不动', () => { const { store, projectId } = setup('auto-easy', 5); const easy = store.createTask({ projectId, title: 'small', complexity: 'easy' }); const medium = store.createTask({ projectId, title: 'mid', complexity: 'medium' }); @@ -85,392 +190,181 @@ test('autonomy=auto-easy:只领 easy,medium ready 不动', async () => { store.decide(medium.id, 'accept', 'user'); // medium → ready assert.equal(store.getTask(medium.id)!.status, 'ready'); - const ran: string[] = []; - const orch = createOrchestrator(store, noopLog, mockDeps({ - runner: async (task) => { ran.push(task.id); return okRun; }, - })); - orch.tick(); - await orch.drain(); + const state = freshState(); + const { deps } = mockDeps(state); + createOrchestrator(store, noopLog, deps).claimTick(); - assert.deepEqual(ran, [easy.id]); - assert.equal(store.getTask(easy.id)!.status, 'exec_review'); - assert.equal(store.getTask(medium.id)!.status, 'ready'); // 不自动跑 + assert.equal(state.spawned.length, 1); + assert.equal(store.getTask(easy.id)!.status, 'executing'); + assert.equal(store.getTask(medium.id)!.status, 'ready'); // 不领 medium store.close(); }); -test('autonomy=auto-approved:领全部 ready(含 medium),依赖未满足/非叶子不领', async () => { +test('auto-approved:领 ready(含 medium);依赖未满足/非叶子不领', () => { const { store, projectId } = setup('auto-approved', 5); const medium = store.createTask({ projectId, title: 'mid', complexity: 'medium' }); store.setSpec(medium.id, '方案'); store.transition(medium.id, 'spec_review'); store.decide(medium.id, 'accept', 'user'); - // 依赖未 done 的任务:建在 medium 上 → blocked,不可领 const dep = store.createTask({ projectId, title: 'after-mid', complexity: 'easy', deps: [medium.id] }); assert.equal(store.getTask(dep.id)!.status, 'blocked'); - const ran: string[] = []; - const orch = createOrchestrator(store, noopLog, mockDeps({ - runner: async (task) => { ran.push(task.id); return okRun; }, - })); - orch.tick(); - await orch.drain(); + const state = freshState(); + const { deps } = mockDeps(state); + createOrchestrator(store, noopLog, deps).claimTick(); - assert.deepEqual(ran, [medium.id]); - assert.equal(store.getTask(medium.id)!.status, 'exec_review'); + assert.equal(state.spawned.length, 1); + assert.equal(store.getTask(medium.id)!.status, 'executing'); assert.equal(store.getTask(dep.id)!.status, 'blocked'); store.close(); }); -test('concurrency=1:同项目同轮只领 1 个,跑完下一轮再领', async () => { - const { store, projectId } = setup('auto-approved', 1); - const t1 = store.createTask({ projectId, title: 'a', complexity: 'easy' }); - const t2 = store.createTask({ projectId, title: 'b', complexity: 'easy' }); - - const gate = deferred(); - const started: string[] = []; - const orch = createOrchestrator(store, noopLog, mockDeps({ - runner: async (task) => { started.push(task.id); await gate.promise; return okRun; }, - })); - - orch.tick(); - await settle(); - assert.deepEqual(started, [t1.id], '并发=1 只应启动第一个任务'); - assert.equal(store.getTask(t1.id)!.status, 'executing'); - assert.equal(store.getTask(t2.id)!.status, 'ready'); - assert.equal(orch.inflight.size, 1); - - orch.tick(); // 在途占满 → 本轮不领 - await settle(); - assert.deepEqual(started, [t1.id]); - - gate.resolve(); - await orch.drain(); - assert.equal(store.getTask(t1.id)!.status, 'exec_review'); - - orch.tick(); // 槽位释放 → 领第二个 - gate.resolve(); - await orch.drain(); - assert.deepEqual(started, [t1.id, t2.id]); - assert.equal(store.getTask(t2.id)!.status, 'exec_review'); - store.close(); -}); - -test('成功路径:状态流转 + setResult(四字段) + executor/reviewer/security 三 run succeeded', async () => { - const { store, projectId } = setup('auto-easy'); - const t = store.createTask({ projectId, title: 'tweak', complexity: 'easy' }); - store.setOperations(t.id, '在 README.md 追加一行'); - - const seen: string[] = []; - store.subscribe((e) => { if (e.type === 'status.changed' && e.taskId === t.id) seen.push(String(e.payload.to)); }); - - let codeReportSeen: string | null = null; - let secReportSeen: string | null = null; - const orch = createOrchestrator(store, noopLog, mockDeps({ - reviewCode: async (_task, _project, _wt, _runId, executorReport) => { - codeReportSeen = executorReport; - return okReview; - }, - reviewSecurity: async (_task, _project, _wt, _runId, executorReport) => { - secReportSeen = executorReport; - return okSecurity; - }, - })); - orch.tick(); - await orch.drain(); - - const done = store.getTask(t.id)!; - assert.equal(done.status, 'exec_review'); - assert.deepEqual(seen, ['queued', 'executing', 'exec_review']); - assert.deepEqual(done.result, { - branch: `maestro/${t.id}`, - worktree: `/tmp/fake-wt/${t.id}`, - diffSummary: ' README.md | 1 +', - commits: ['abc1234 hello maestro'], - prUrl: null, - summary: '## 做了什么\nmock 复审通过', - verdict: 'approve', - securitySummary: '## 安全审计\nmock 审计通过', - securityVerdict: 'approve', - mergeTaskId: null, - }); - assert.equal(codeReportSeen, '执行自述:改了 README'); // runner finalText 传给两个复审作执行者自述 - assert.equal(secReportSeen, '执行自述:改了 README'); - - const runs = store.listRuns(t.id); - assert.equal(runs.length, 3); - const executor = runs.find((r) => r.kind === 'executor')!; - assert.equal(executor.status, 'succeeded'); - assert.equal(executor.branch, `maestro/${t.id}`); - assert.equal(executor.transcriptRef, '/tmp/fake.jsonl'); - assert.equal(executor.claudeSessionId, 'sess-mock-1'); - const reviewer = runs.find((r) => r.kind === 'reviewer')!; - assert.equal(reviewer.status, 'succeeded'); - assert.equal(reviewer.transcriptRef, '/tmp/fake-review.jsonl'); - assert.equal(reviewer.claudeSessionId, 'sess-review-1'); - const security = runs.find((r) => r.kind === 'security')!; - assert.equal(security.status, 'succeeded'); - assert.equal(security.transcriptRef, '/tmp/fake-security.jsonl'); - assert.equal(security.claudeSessionId, 'sess-security-1'); - store.close(); -}); - -test('任一 verdict=reject 也照常落进 result(最终裁决仍归用户)', async () => { - const { store, projectId } = setup('auto-easy'); - const t = store.createTask({ projectId, title: 'risky', complexity: 'easy' }); - const orch = createOrchestrator(store, noopLog, mockDeps({ - reviewCode: async () => ({ ...okReview, summary: '发现问题', verdict: 'reject' as const }), - reviewSecurity: async () => ({ ...okSecurity, summary: '发现密钥泄露', verdict: 'reject' as const }), - })); - orch.tick(); - await orch.drain(); - const done = store.getTask(t.id)!; - assert.equal(done.status, 'exec_review'); - assert.equal(done.result!.verdict, 'reject'); - assert.equal(done.result!.summary, '发现问题'); - assert.equal(done.result!.securityVerdict, 'reject'); - assert.equal(done.result!.securitySummary, '发现密钥泄露'); - store.close(); -}); - -test('code review 失败不挡任务:summary 记失败原因、verdict=null,安全审计照常跑', async () => { - const { store, projectId } = setup('auto-easy'); - const t = store.createTask({ projectId, title: 'review-broken', complexity: 'easy' }); - const orch = createOrchestrator(store, noopLog, mockDeps({ - reviewCode: async () => { throw new Error('复审 CC 崩了'); }, - })); - orch.tick(); - await orch.drain(); - - const done = store.getTask(t.id)!; - assert.equal(done.status, 'exec_review'); // 不挡结果闸 - assert.equal(done.result!.verdict, null); - assert.equal(done.result!.summary, '自动复审失败:复审 CC 崩了'); - assert.equal(done.result!.securityVerdict, 'approve'); // 另一个复审不受影响 - assert.equal(done.result!.securitySummary, '## 安全审计\nmock 审计通过'); - - const runs = store.listRuns(t.id); - assert.equal(runs.find((r) => r.kind === 'executor')!.status, 'succeeded'); - const reviewer = runs.find((r) => r.kind === 'reviewer')!; - assert.equal(reviewer.status, 'failed'); - assert.match(reviewer.error ?? '', /复审 CC 崩了/); - assert.equal(runs.find((r) => r.kind === 'security')!.status, 'succeeded'); - store.close(); -}); - -test('安全审计失败不挡任务:securitySummary 记失败原因、securityVerdict=null,code review 不受影响', async () => { - const { store, projectId } = setup('auto-easy'); - const t = store.createTask({ projectId, title: 'security-broken', complexity: 'easy' }); - const orch = createOrchestrator(store, noopLog, mockDeps({ - reviewSecurity: async () => { throw new Error('审计 CC 崩了'); }, - })); - orch.tick(); - await orch.drain(); - - const done = store.getTask(t.id)!; - assert.equal(done.status, 'exec_review'); - assert.equal(done.result!.verdict, 'approve'); - assert.equal(done.result!.summary, '## 做了什么\nmock 复审通过'); - assert.equal(done.result!.securityVerdict, null); - assert.equal(done.result!.securitySummary, '自动复审失败:审计 CC 崩了'); - - const runs = store.listRuns(t.id); - assert.equal(runs.find((r) => r.kind === 'executor')!.status, 'succeeded'); - assert.equal(runs.find((r) => r.kind === 'reviewer')!.status, 'succeeded'); - const security = runs.find((r) => r.kind === 'security')!; - assert.equal(security.status, 'failed'); - assert.match(security.error ?? '', /审计 CC 崩了/); - store.close(); -}); - -test('verify 不过:按失败处理(run failed + 重新入队)', async () => { - const { store, projectId } = setup('auto-easy'); - const t = store.createTask({ projectId, title: 'v', complexity: 'easy' }); - const orch = createOrchestrator(store, noopLog, mockDeps({ - verify: async () => ({ ok: false, exitCode: 1, logRef: '/tmp/v.log', error: 'verify 失败(exit 1)' }), - })); - orch.tick(); - await orch.drain(); - assert.equal(store.getTask(t.id)!.status, 'queued'); // 第一次失败 → 重新入队 - const runs = store.listRuns(t.id); - assert.equal(runs.length, 1); - assert.equal(runs[0].status, 'failed'); - assert.match(runs[0].error ?? '', /verify 失败/); - store.close(); -}); - -test(`失败重试:重试 ${DEFAULT_MAX_RETRIES} 次后 → needs_attention(共 ${DEFAULT_MAX_RETRIES + 1} 次失败 run)`, async () => { - const { store, projectId } = setup('auto-easy'); - const t = store.createTask({ projectId, title: 'flaky', complexity: 'easy' }); - - let attempts = 0; - // 使用快进时钟跳过退避冷却(每次 tick 前将时钟推进 10min) - const clock = makeClock(); - const orch = createOrchestrator(store, noopLog, mockDeps({ - runner: async () => { attempts++; return { ok: false, transcriptRef: null, sessionId: null, error: `boom #${attempts}` }; }, - nowMs: clock.nowMs, - })); - - for (let i = 1; i <= DEFAULT_MAX_RETRIES; i++) { - clock.advance(10 * 60_000); // 跳过退避冷却 - orch.tick(); - await orch.drain(); - assert.equal(store.getTask(t.id)!.status, 'queued', `第 ${i} 次失败后应重新入队`); - } - - clock.advance(10 * 60_000); // 跳过最后一次退避 - orch.tick(); // 最后一次重试也失败 - await orch.drain(); - assert.equal(store.getTask(t.id)!.status, 'needs_attention'); - assert.equal(attempts, DEFAULT_MAX_RETRIES + 1); - - const failed = store.listRuns(t.id).filter((r) => r.status === 'failed'); - assert.equal(failed.length, DEFAULT_MAX_RETRIES + 1); - - clock.advance(10 * 60_000); - orch.tick(); // needs_attention 不会再被领取 - await orch.drain(); - assert.equal(attempts, DEFAULT_MAX_RETRIES + 1); - store.close(); -}); - -test('项目级 maxRetries=1:只重试 1 次就 → needs_attention', async () => { - const store = new Store(':memory:'); - const p = store.createProject({ - name: 'orch', repoPath: '/tmp/orch-repo-retries-' + Math.random(), - autonomy: 'auto-easy', maxRetries: 1, - }); - const projectId = p.id; - const t = store.createTask({ projectId, title: 'fail1', complexity: 'easy' }); - - let attempts = 0; - const clock = makeClock(); - const orch = createOrchestrator(store, noopLog, mockDeps({ - runner: async () => { attempts++; return { ok: false, transcriptRef: null, sessionId: null, error: 'boom' }; }, - nowMs: clock.nowMs, - })); - - // 第 1 次执行失败 → 重新入队(还有 1 次重试机会) - orch.tick(); - await orch.drain(); - assert.equal(store.getTask(t.id)!.status, 'queued'); - - // 第 2 次失败 → 超过 maxRetries=1 → needs_attention(快进时钟跳过退避) - clock.advance(10 * 60_000); - orch.tick(); - await orch.drain(); - assert.equal(store.getTask(t.id)!.status, 'needs_attention'); - assert.equal(attempts, 2); - store.close(); -}); - -test('项目级 maxRetries=0:首次失败直接 needs_attention(不重试)', async () => { - const store = new Store(':memory:'); - const p = store.createProject({ - name: 'orch', repoPath: '/tmp/orch-repo-zero-' + Math.random(), - autonomy: 'auto-easy', maxRetries: 0, - }); - const t = store.createTask({ projectId: p.id, title: 'fail0', complexity: 'easy' }); - - let attempts = 0; - const orch = createOrchestrator(store, noopLog, mockDeps({ - runner: async () => { attempts++; return { ok: false, transcriptRef: null, sessionId: null, error: 'boom' }; }, - })); - - orch.tick(); - await orch.drain(); - assert.equal(store.getTask(t.id)!.status, 'needs_attention'); - assert.equal(attempts, 1); - store.close(); -}); - -test('computeBackoffMs:指数退避公式,30s 基数,上限 10min', () => { - assert.equal(computeBackoffMs(1), 30_000); - assert.equal(computeBackoffMs(2), 60_000); - assert.equal(computeBackoffMs(3), 120_000); - assert.equal(computeBackoffMs(4), 240_000); - assert.equal(computeBackoffMs(10), 10 * 60_000); // capped at 10min - assert.equal(computeBackoffMs(100), 10 * 60_000); -}); - -test('退避冷却:首次失败重入队后,退避期内不被领取;退避过期后正常领取', async () => { +test('退避:nextEligibleAt 在未来 → 不领;过期后 → 领', () => { const { store, projectId } = setup('auto-easy'); const t = store.createTask({ projectId, title: 'backoff', complexity: 'easy' }); + // 任务已在 queued 且退避到未来 + store.transition(t.id, 'queued', { by: 'test' }); + const future = new Date(Date.now() + 60_000).toISOString(); + store.setNextEligibleAt(t.id, future); - let attempts = 0; - // 使用可控时钟:初始时间 0,退避 30s(attempt=1 → BACKOFF_BASE=30000ms) - const clock = makeClock(0); - const orch = createOrchestrator(store, noopLog, mockDeps({ - runner: async () => { attempts++; return { ok: false, transcriptRef: null, sessionId: null, error: 'boom' }; }, - nowMs: clock.nowMs, - })); + const clock = makeClock(Date.now()); + const state = freshState(); + const { deps } = mockDeps(state, { nowMs: clock.nowMs }); + const orch = createOrchestrator(store, noopLog, deps); - // 第 1 次失败(t=0)→ 重入队 + 退避(backoffUntil = 0 + 30000 = 30000ms) - orch.tick(); - await orch.drain(); - assert.equal(store.getTask(t.id)!.status, 'queued'); - assert.equal(attempts, 1); - - // t=29999:退避期内,不被领取 - clock.advance(29_999); - orch.tick(); - await orch.drain(); - assert.equal(attempts, 1, '退避期内不应再次执行'); + orch.claimTick(); + assert.equal(state.spawned.length, 0, '退避期内不领'); assert.equal(store.getTask(t.id)!.status, 'queued'); - // t=30001:退避过期,任务应被重新领取 - clock.advance(2); - orch.tick(); - await orch.drain(); - assert.equal(attempts, 2, '退避过期后应再次执行'); + clock.advance(61_000); // 退避过期 + orch.claimTick(); + assert.equal(state.spawned.length, 1, '退避过期后领取'); + assert.equal(store.getTask(t.id)!.status, 'executing'); store.close(); }); -test('requeueTask:needs_attention → queued,重置重试基线,再次允许重试', async () => { +test('spawnWorker 抛错:兜底 failTaskAttempt(task 转 queued,executor run failed)', () => { const { store, projectId } = setup('auto-easy'); - const t = store.createTask({ projectId, title: 'requeue', complexity: 'easy' }); + const t = store.createTask({ projectId, title: 'boom-spawn', complexity: 'easy' }); + const state = freshState(); + const { deps } = mockDeps(state, { + spawnWorker: () => { throw new Error('spawn 失败:ENOENT'); }, + }); + createOrchestrator(store, noopLog, deps).claimTick(); - let attempts = 0; - const clock = makeClock(); - const orch = createOrchestrator(store, noopLog, mockDeps({ - runner: async () => { attempts++; return { ok: false, transcriptRef: null, sessionId: null, error: 'boom' }; }, - nowMs: clock.nowMs, - })); + const after = store.getTask(t.id)!; + assert.equal(after.status, 'queued', '默认 maxRetries=2,首次失败 → 重入队'); + assert.ok(after.nextEligibleAt, '退避已持久化'); + const executor = store.listRuns(t.id).find((r) => r.kind === 'executor')!; + assert.equal(executor.status, 'failed'); + assert.match(executor.error ?? '', /spawn 失败/); + store.close(); +}); - // 耗尽默认重试次数 → needs_attention(每次需快进时钟跳过退避) - for (let i = 0; i <= DEFAULT_MAX_RETRIES; i++) { - clock.advance(10 * 60_000); - orch.tick(); - await orch.drain(); - } +// ───────────────────────── reaper(回收死 worker)───────────────────────── + +test('reaper:isWorkerAlive=false → failTaskAttempt(executing→queued,executor run failed)', () => { + const { store, projectId } = setup('auto-easy'); + const t = store.createTask({ projectId, title: 'dead', complexity: 'easy' }); + // 先让它进 executing + 建 run(用一次正常领取,alive=true 不会被回收) + const state = freshState(); + const { deps } = mockDeps(state); + const orch = createOrchestrator(store, noopLog, deps); + orch.claimTick(); + assert.equal(store.getTask(t.id)!.status, 'executing'); + const runId = store.listRuns(t.id)[0].id; + + // 标记 worker 已死 → reaper 回收 + state.alive = false; + orch.reap(); + + const after = store.getTask(t.id)!; + assert.equal(after.status, 'queued', '默认 maxRetries=2,首次回收 → 重入队'); + assert.ok(after.nextEligibleAt, '退避已持久化'); + const executor = store.getRun(runId)!; + assert.equal(executor.status, 'failed'); + assert.match(executor.error ?? '', /worker 异常退出/); + store.close(); +}); + +test('reaper:isWorkerAlive=true → 不回收(保持 executing)', () => { + const { store, projectId } = setup('auto-easy'); + const t = store.createTask({ projectId, title: 'alive', complexity: 'easy' }); + const state = freshState(); + const { deps } = mockDeps(state); + const orch = createOrchestrator(store, noopLog, deps); + orch.claimTick(); + assert.equal(store.getTask(t.id)!.status, 'executing'); + + state.alive = true; + orch.reap(); + assert.equal(store.getTask(t.id)!.status, 'executing', 'worker 活 → 不回收'); + store.close(); +}); + +test('reaper:maxRetries=0 → 死 worker 直接 needs_attention', () => { + const store = new Store(':memory:'); + const p = store.createProject({ + name: 'orch', repoPath: '/tmp/orch-reap0-' + Math.random(), autonomy: 'auto-easy', maxRetries: 0, + }); + const t = store.createTask({ projectId: p.id, title: 'x', complexity: 'easy' }); + const state = freshState(); + const { deps } = mockDeps(state); + const orch = createOrchestrator(store, noopLog, deps); + orch.claimTick(); + state.alive = false; + orch.reap(); assert.equal(store.getTask(t.id)!.status, 'needs_attention'); - assert.equal(attempts, DEFAULT_MAX_RETRIES + 1); + store.close(); +}); - // 手动重投:重置基线,转 queued - const requeued = store.requeueTask(t.id); - assert.equal(requeued.status, 'queued'); - assert.equal(requeued.retryBaseline, DEFAULT_MAX_RETRIES + 1); +// ───────────────────────── 多轮:reaper 腾槽后重领(退避到期)───────────────────────── - // 重投后编排器应能再次执行(快进时钟确保无退避阻拦) +test('死 worker 回收 + 退避到期后下一轮重新领取(监工闭环)', () => { + const { store, projectId } = setup('auto-easy', 1); + const t = store.createTask({ projectId, title: 'recycle', complexity: 'easy' }); + const clock = makeClock(Date.now()); + const state = freshState(); + const { deps } = mockDeps(state, { nowMs: clock.nowMs }); + const orch = createOrchestrator(store, noopLog, deps); + + // 第一轮:领取 + spawn(alive=true) + orch.tick(); + assert.equal(state.spawned.length, 1); + assert.equal(store.getTask(t.id)!.status, 'executing'); + + // worker 死 → 下一轮 reaper 回收(→ queued + 退避),退避期内不重领 + state.alive = false; + orch.tick(); + assert.equal(store.getTask(t.id)!.status, 'queued'); + assert.equal(state.spawned.length, 1, '退避期内不重领'); + + // 退避到期 → 再下一轮重新领取 clock.advance(10 * 60_000); orch.tick(); - await orch.drain(); - assert.equal(attempts, DEFAULT_MAX_RETRIES + 2, '重投后应再次执行'); - // 第 1 次净失败后应重入队(基线已重置,净失败=1 < maxRetries=2) - assert.equal(store.getTask(t.id)!.status, 'queued'); + assert.equal(state.spawned.length, 2, '退避到期 → 重新领取 spawn'); + assert.equal(store.getTask(t.id)!.status, 'executing'); store.close(); }); -test('project paused:不领取', async () => { - const { store, projectId } = setup('auto-approved'); - const t = store.createTask({ projectId, title: 'x', complexity: 'easy' }); - store.patchProject(projectId, { status: 'paused' }); - let calls = 0; - const orch = createOrchestrator(store, noopLog, mockDeps({ - runner: async () => { calls++; return okRun; }, - })); - orch.tick(); - await orch.drain(); - assert.equal(calls, 0); - assert.equal(store.getTask(t.id)!.status, 'ready'); +test(`reaper 反复回收:净失败 ${DEFAULT_MAX_RETRIES} 次后 → needs_attention`, () => { + const { store, projectId } = setup('auto-easy', 1); + const t = store.createTask({ projectId, title: 'flaky', complexity: 'easy' }); + const clock = makeClock(Date.now()); + const state = freshState(); + const { deps } = mockDeps(state, { nowMs: clock.nowMs }); + const orch = createOrchestrator(store, noopLog, deps); + + // 反复:领取(alive=true 领取那刻)→ 标死 → reaper 回收 + for (let i = 0; i <= DEFAULT_MAX_RETRIES; i++) { + clock.advance(10 * 60_000); // 跳过退避 + state.alive = true; + orch.claimTick(); + state.alive = false; + orch.reap(); + } + assert.equal(store.getTask(t.id)!.status, 'needs_attention'); + const failed = store.listRuns(t.id).filter((r) => r.kind === 'executor' && r.status === 'failed'); + assert.equal(failed.length, DEFAULT_MAX_RETRIES + 1); store.close(); });