merge phase3: 多进程执行解耦(daemon 监工 + worker 进程 + 文件通讯 + 重启 re-adopt)
This commit is contained in:
wangjia
+11
-2
@@ -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<void> {
|
||||
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/ 静态文件）
|
||||
@@ -62,7 +71,7 @@ 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 标记中断）`);
|
||||
if (itr.readopted || itr.reclaimed) app.log.info(`中断恢复：re-adopt ${itr.readopted} · 回收 ${itr.reclaimed}`);
|
||||
|
||||
const syncTimer = startSyncLoop(store, app);
|
||||
const orchTimer = startOrchestrator(store, app); // 编排器：自动领取可执行任务（MAESTRO_ORCH_INTERVAL 秒，0=关闭）
|
||||
|
||||
@@ -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/<runId>/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);
|
||||
}
|
||||
}
|
||||
+144
-168
@@ -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<WorktreeInfo>;
|
||||
worktreeDiff: (repoPath: string, dir: string, branch: string, baseBranch: string) => Promise<WorktreeDiff>;
|
||||
/** 当前时间戳（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<void>;
|
||||
/** 在途任务 id（防同任务重复领取） */
|
||||
readonly inflight: ReadonlyMap<string, string>;
|
||||
/** 仅领取/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<OrchestratorDeps> = {}): Orchestrator {
|
||||
const d: OrchestratorDeps = { runner: runTask, reviewCode, reviewSecurity, verify: runVerify, createWorktree, worktreeDiff, nowMs: Date.now, ...deps };
|
||||
const inflight = new Map<string, string>(); // taskId → projectId
|
||||
const pending = new Set<Promise<void>>();
|
||||
/** 退避表：taskId → 下次可领取时间戳（ms）；daemon 重启后清空，已有 queued 任务即刻可领 */
|
||||
const backoffUntil = new Map<string, number>();
|
||||
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<void> {
|
||||
/** 领取一个任务：建分支/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<concurrency 时按 claimable 领新任务并 spawn worker。 */
|
||||
function claimTick(): 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++;
|
||||
let active = store.countExecuting(p.id);
|
||||
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);
|
||||
claimOne(p, task, score);
|
||||
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}`);
|
||||
log.error(`编排器领取轮失败：${(e as Error).message}`);
|
||||
}
|
||||
}
|
||||
|
||||
async function drain(): Promise<void> {
|
||||
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;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
// Phase 3 多进程执行：worker 侧【纯执行管线】。
|
||||
//
|
||||
// 这是旧 `orchestrator.executeTask` 函数体「劈出来」的执行半边——只负责跑
|
||||
// worktree→执行→verify→双复审→产出，并把每一步的进度/结果经 `emit` 写进 outbox 文件。
|
||||
// 它【完全不碰 DB】（连读都不碰），也【不做任何重试 / needs_attention 决策】：
|
||||
// - 失败（executor 报错 / verify 不过）→ emit failed，然后 emit done，结束（不抛错）。
|
||||
// - 成功 → emit result（四要素），然后 emit done。
|
||||
// 重试次数、退避、转终态等判断全部归 daemon（它读 outbox 后决策）。
|
||||
//
|
||||
// daemon 侧的 store.* 调用与本模块 emit 的对应关系（仅为说明，本模块不引用 store）：
|
||||
// store.transition(executing) / startRun(executor) → emit started（由 worker 在入口发，不在此）
|
||||
// runner 失败 / verify 失败 finishRun(failed) → emit failed
|
||||
// 双复审 + setResult(四字段) + transition(exec_review) → emit result
|
||||
// （worker 退出） → emit done
|
||||
|
||||
import type { JobSpec, OutboxPayload, ReviewReport } from './protocol.js';
|
||||
import { createWorktree, worktreeDiff, type WorktreeDiff, type WorktreeInfo } from './worktree.js';
|
||||
import { runTask, type RunnerFn } from './runner.js';
|
||||
import { runVerify, type VerifyFn } from './verify.js';
|
||||
import { reviewCode, reviewSecurity, type ReviewerFn } from './reviewer.js';
|
||||
|
||||
/** pipeline 的依赖注入面（便于单测 mock）；默认值取真实现。 */
|
||||
export interface PipelineDeps {
|
||||
createWorktree: (repoPath: string, taskId: string, baseBranch: string) => Promise<WorktreeInfo>;
|
||||
worktreeDiff: (repoPath: string, dir: string, branch: string, baseBranch: string) => Promise<WorktreeDiff>;
|
||||
runTask: RunnerFn;
|
||||
verify: VerifyFn;
|
||||
reviewCode: ReviewerFn; // code review（daemon 落成 kind=reviewer 的 run）
|
||||
reviewSecurity: ReviewerFn; // 安全审计（daemon 落成 kind=security 的 run）
|
||||
}
|
||||
|
||||
/** 真实依赖（worker 生产用）。 */
|
||||
export const realDeps: PipelineDeps = {
|
||||
createWorktree,
|
||||
worktreeDiff,
|
||||
runTask,
|
||||
verify: runVerify,
|
||||
reviewCode,
|
||||
reviewSecurity,
|
||||
};
|
||||
|
||||
/** emit：把一条 OutboxPayload 交给 outbox（worker 传 appendOutbox 包装，测试传收集器）。 */
|
||||
export type Emit = (payload: OutboxPayload) => void;
|
||||
|
||||
/**
|
||||
* 跑一个独立复审（code / security），复刻旧 `runOneReview` 的兜底：
|
||||
* 任何失败折叠成 `{summary:'自动复审失败：'+msg, verdict:null, transcriptRef:null}`，不抛错、不挡任务。
|
||||
* 成功时把 ReviewResult 映射成 ReviewReport（丢弃 sessionId——result 记录里复审只留 transcriptRef）。
|
||||
*/
|
||||
async function runOneReview(
|
||||
fn: ReviewerFn,
|
||||
job: JobSpec,
|
||||
wt: WorktreeInfo,
|
||||
reviewRunId: string,
|
||||
executorReport: string,
|
||||
): Promise<ReviewReport> {
|
||||
try {
|
||||
const rv = await fn(job.task, job.project, wt, reviewRunId, executorReport);
|
||||
return { summary: rv.summary, verdict: rv.verdict, transcriptRef: rv.transcriptRef };
|
||||
} catch (e) {
|
||||
return { summary: `自动复审失败：${(e as Error).message}`, verdict: null, transcriptRef: null };
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 纯执行管线。复刻旧 in-process 流程的【执行与产出】，但所有 store.* 改成 emit(outbox)：
|
||||
* 1. createWorktree
|
||||
* 2. runTask；!ok → emit failed{error,transcriptRef,sessionId} + done，返回（不抛）
|
||||
* 3. runVerify；!ok → emit failed{error,rr.transcriptRef,rr.sessionId} + done，返回
|
||||
* 4. worktreeDiff
|
||||
* 5. 双复审（顺序，任一失败不挡，见 runOneReview 兜底）
|
||||
* 6. emit result（branch/worktree/diffSummary/commits/executor/code/security）+ done
|
||||
*
|
||||
* 不抛错（除非 emit 自身抛——那由 worker 顶层兜底）。全程总会 emit 一条终态（failed|result）后再 emit done。
|
||||
*/
|
||||
export async function runPipeline(job: JobSpec, deps: PipelineDeps = realDeps, emit: Emit): Promise<void> {
|
||||
// 1. 建 worktree（其 dir/branch 应与 job.worktreeDir/branch 一致；以 deps 真建的为准）
|
||||
const wt = await deps.createWorktree(job.project.repoPath, job.task.id, job.project.defaultBranch);
|
||||
|
||||
// 2. 执行
|
||||
emit({ type: 'phase', phase: 'executing' });
|
||||
const rr = await deps.runTask(job.task, job.project, wt, job.runId);
|
||||
if (!rr.ok) {
|
||||
emit({ type: 'failed', error: rr.error ?? '执行失败', transcriptRef: rr.transcriptRef, sessionId: rr.sessionId });
|
||||
emit({ type: 'done' });
|
||||
return;
|
||||
}
|
||||
|
||||
// 3. 校验
|
||||
emit({ type: 'phase', phase: 'verifying' });
|
||||
const vr = await deps.verify(job.project, wt.dir, job.runId);
|
||||
if (!vr.ok) {
|
||||
// transcriptRef/sessionId 沿用 executor 的（verify 自身无转录）
|
||||
emit({ type: 'failed', error: vr.error ?? 'verify 失败', transcriptRef: rr.transcriptRef, sessionId: rr.sessionId });
|
||||
emit({ type: 'done' });
|
||||
return;
|
||||
}
|
||||
|
||||
// 4. diff
|
||||
const diff = await deps.worktreeDiff(job.project.repoPath, wt.dir, wt.branch, job.project.defaultBranch);
|
||||
|
||||
// 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);
|
||||
|
||||
// 6. 成功终态
|
||||
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' });
|
||||
}
|
||||
@@ -0,0 +1,162 @@
|
||||
// Worker ↔ daemon 契约（Phase 3 多进程执行）。
|
||||
//
|
||||
// 铁律：worker 进程【完全不碰 DB】（连读都不碰）。daemon 是唯一 DB 读写者。
|
||||
// 二者只经【文件 + 进程信号】通讯，天然跨 daemon 重启持久：
|
||||
// daemon → worker：runs/<runId>/job.json（唯一输入）；SIGTERM（取消/超时）
|
||||
// worker → daemon：runs/<runId>/outbox.ndjson（追加、带 seq）；runs/<runId>/heartbeat（mtime 心跳）
|
||||
//
|
||||
// 本模块同时被 worker 侧（写 outbox/读 job/刷心跳）与 daemon 侧（写 job/读 outbox/判活）import，
|
||||
// 是唯一的共享面——双方都不应另起自己的协议常量/路径计算。
|
||||
|
||||
import {
|
||||
existsSync, mkdirSync, readFileSync, writeFileSync, appendFileSync,
|
||||
statSync, utimesSync, closeSync, openSync,
|
||||
} from 'node:fs';
|
||||
import { homedir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
import type { Project, Task, ReviewVerdict } from '../model/types.js';
|
||||
|
||||
/** 数据根目录（与 worktree.ts 同约定）：<MAESTRO_DATA_DIR 或 ~/.maestro> */
|
||||
function dataDir(): string {
|
||||
return process.env.MAESTRO_DATA_DIR ?? join(homedir(), '.maestro');
|
||||
}
|
||||
|
||||
/** 全部 run 工作目录根：<data>/runs */
|
||||
export function runsBase(): string {
|
||||
return join(dataDir(), 'runs');
|
||||
}
|
||||
/** 单个 run 的工作目录：<runsBase>/<runId>（放 job.json / outbox.ndjson / heartbeat） */
|
||||
export function runDir(runId: string): string {
|
||||
return join(runsBase(), runId);
|
||||
}
|
||||
export function jobPath(runId: string): string { return join(runDir(runId), 'job.json'); }
|
||||
export function outboxPath(runId: string): string { return join(runDir(runId), 'outbox.ndjson'); }
|
||||
export function heartbeatPath(runId: string): string { return join(runDir(runId), 'heartbeat'); }
|
||||
|
||||
// ───────────────────────── daemon → worker：JobSpec ─────────────────────────
|
||||
|
||||
/**
|
||||
* worker 的唯一输入。daemon 在 spawn 前写 runs/<runId>/job.json，worker 读它即可执行，
|
||||
* 【无需访问 DB】。携带完整 Task / Project（均 JSON 可序列化），worker 自行 pickModel/buildPrompt。
|
||||
*/
|
||||
export interface JobSpec {
|
||||
runId: string;
|
||||
task: Task;
|
||||
project: Project;
|
||||
/** 确定性 worktree 路径与分支（daemon 预填；worker 也能由 worktree.ts 算出，传入避免重复） */
|
||||
worktreeDir: string;
|
||||
branch: string;
|
||||
}
|
||||
|
||||
export function writeJobSpec(job: JobSpec): void {
|
||||
mkdirSync(runDir(job.runId), { recursive: true });
|
||||
writeFileSync(jobPath(job.runId), JSON.stringify(job));
|
||||
}
|
||||
export function readJobSpec(runId: string): JobSpec {
|
||||
return JSON.parse(readFileSync(jobPath(runId), 'utf8')) as JobSpec;
|
||||
}
|
||||
|
||||
// ───────────────────────── worker → daemon：OutboxRecord ────────────────────
|
||||
|
||||
/** 一次复审（code review / 安全审计）的产出，worker 报给 daemon，由 daemon 落成 run 行 + result 字段 */
|
||||
export interface ReviewReport {
|
||||
summary: string | null;
|
||||
verdict: ReviewVerdict | null;
|
||||
transcriptRef: string | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* worker 追加进 outbox 的记录。`seq` 单调递增（每 run 内），`at` ISO 时间，均由 appendOutbox 填。
|
||||
* - started：worker 启动，报自己的 pid / worktree / branch / model
|
||||
* - phase ：进度阶段（executing|verifying|reviewing…），daemon 落成 event 推看板（可选）
|
||||
* - failed ：本次尝试终态失败（executor 报错或 verify 失败）。daemon 据此 finishRun(failed)+重试决策
|
||||
* - result ：成功终态。daemon 据此建 reviewer/security run + setResult(四字段) + 转 exec_review
|
||||
* - done ：worker 即将退出（成功或失败都发，daemon 据此停止 tail）
|
||||
*/
|
||||
export type OutboxRecord =
|
||||
| { seq: number; at: string; type: 'started'; pid: number; worktree: string; branch: string; model: string }
|
||||
| { seq: number; at: string; type: 'phase'; phase: string }
|
||||
| { seq: number; at: string; type: 'failed'; error: string; transcriptRef: string | null; sessionId: string | null }
|
||||
| {
|
||||
seq: number; at: string; type: 'result';
|
||||
branch: string; worktree: string; diffSummary: string; commits: string[];
|
||||
executor: { transcriptRef: string | null; sessionId: string | null };
|
||||
code: ReviewReport;
|
||||
security: ReviewReport;
|
||||
}
|
||||
| { seq: number; at: string; type: 'done' };
|
||||
|
||||
/** OutboxRecord 去掉 seq/at（由 appendOutbox 填） */
|
||||
export type OutboxPayload =
|
||||
| Omit<Extract<OutboxRecord, { type: 'started' }>, 'seq' | 'at'>
|
||||
| Omit<Extract<OutboxRecord, { type: 'phase' }>, 'seq' | 'at'>
|
||||
| Omit<Extract<OutboxRecord, { type: 'failed' }>, 'seq' | 'at'>
|
||||
| Omit<Extract<OutboxRecord, { type: 'result' }>, 'seq' | 'at'>
|
||||
| Omit<Extract<OutboxRecord, { type: 'done' }>, 'seq' | 'at'>;
|
||||
|
||||
/** 追加一条 outbox 记录（worker 侧调用）。seq = 现有行数+1（worker 单线程，无并发写）。返回写入的完整记录。 */
|
||||
export function appendOutbox(runId: string, payload: OutboxPayload): OutboxRecord {
|
||||
mkdirSync(runDir(runId), { recursive: true });
|
||||
const prev = readOutboxAll(runId);
|
||||
const seq = prev.length + 1;
|
||||
const rec = { seq, at: new Date().toISOString(), ...payload } as OutboxRecord;
|
||||
appendFileSync(outboxPath(runId), JSON.stringify(rec) + '\n');
|
||||
return rec;
|
||||
}
|
||||
|
||||
/** 读全部 outbox 记录（坏行跳过，容忍写一半的尾行）。 */
|
||||
export function readOutboxAll(runId: string): OutboxRecord[] {
|
||||
const p = outboxPath(runId);
|
||||
if (!existsSync(p)) return [];
|
||||
const out: OutboxRecord[] = [];
|
||||
for (const line of readFileSync(p, 'utf8').split('\n')) {
|
||||
const s = line.trim();
|
||||
if (!s) continue;
|
||||
try { out.push(JSON.parse(s) as OutboxRecord); } catch { /* 半截尾行：跳过，下轮再读 */ }
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/** 读 seq > lastSeq 的新记录（daemon ingest 侧调用，幂等去重靠 lastSeq）。 */
|
||||
export function readOutboxSince(runId: string, lastSeq: number): OutboxRecord[] {
|
||||
return readOutboxAll(runId).filter((r) => r.seq > lastSeq);
|
||||
}
|
||||
|
||||
// ───────────────────────── worker → daemon：心跳 + 存活判定 ──────────────────
|
||||
|
||||
/** worker 刷心跳：touch runs/<runId>/heartbeat（文件不存在则建）。daemon 用其 mtime 判活。 */
|
||||
export function touchHeartbeat(runId: string): void {
|
||||
mkdirSync(runDir(runId), { recursive: true });
|
||||
const p = heartbeatPath(runId);
|
||||
if (!existsSync(p)) { closeSync(openSync(p, 'w')); return; }
|
||||
const now = new Date();
|
||||
utimesSync(p, now, now);
|
||||
}
|
||||
|
||||
/** 心跳文件的「年龄」毫秒（now - mtime）；无心跳文件返回 null。 */
|
||||
export function heartbeatAgeMs(runId: string, now = Date.now()): number | null {
|
||||
try { return now - statSync(heartbeatPath(runId)).mtimeMs; }
|
||||
catch { return null; }
|
||||
}
|
||||
|
||||
/** pid 是否存活（signal 0 探测；EPERM 视为存活——进程在但无权） */
|
||||
export function pidAlive(pid: number): boolean {
|
||||
if (!pid || pid <= 0) return false;
|
||||
try { process.kill(pid, 0); return true; }
|
||||
catch (e) { return (e as NodeJS.ErrnoException).code === 'EPERM'; }
|
||||
}
|
||||
|
||||
export const HEARTBEAT_INTERVAL_MS = 10_000; // worker 刷心跳间隔
|
||||
export const HEARTBEAT_GRACE_MS = 60_000; // 心跳超过此年龄视为可疑（须 > interval，容忍 GC/慢盘）
|
||||
export const BOOT_GRACE_MS = 30_000; // started 后此窗口内额外认 pid 存活（覆盖启动到首次心跳空窗）
|
||||
|
||||
/**
|
||||
* worker 是否存活。主信号＝心跳新鲜；boot 窗口内额外接受 pid 存活（覆盖启动空窗）；
|
||||
* 窗口后仅认心跳——规避 pid 复用误判（被复用的 pid 不会更新本 run 的心跳）。
|
||||
*/
|
||||
export function isWorkerAlive(args: { pid: number | null; heartbeatAgeMs: number | null; startedAgeMs: number }): boolean {
|
||||
const { pid, heartbeatAgeMs: hb, startedAgeMs } = args;
|
||||
if (hb !== null && hb < HEARTBEAT_GRACE_MS) return true; // 心跳新鲜 → 活
|
||||
if (startedAgeMs < BOOT_GRACE_MS && pid !== null && pidAlive(pid)) return true; // 启动空窗 → pid 兜底
|
||||
return false;
|
||||
}
|
||||
@@ -0,0 +1,93 @@
|
||||
// Phase 3 多进程执行：worker 进程【入口】。
|
||||
//
|
||||
// node worker.js <runId> （或 tsx worker.ts <runId>）
|
||||
//
|
||||
// daemon 在 spawn 前写好 runs/<runId>/job.json，本进程读它即可执行，【完全不碰 DB】。
|
||||
// 一切进度/结果只经 protocol 的文件协议汇报：
|
||||
// - 起手 emit started（pid/worktree/branch/model）
|
||||
// - 周期 touchHeartbeat（daemon 据 mtime 判活）
|
||||
// - runPipeline 内部 emit failed|result + done（终态）
|
||||
// - 意外异常 / 收到 SIGTERM → 兜底 emit failed + done，然后退出
|
||||
//
|
||||
// 顶层【不 import 任何 store/db】——worker 与 DB 完全隔离。
|
||||
|
||||
import {
|
||||
readJobSpec,
|
||||
appendOutbox,
|
||||
touchHeartbeat,
|
||||
HEARTBEAT_INTERVAL_MS,
|
||||
type OutboxPayload,
|
||||
} from './protocol.js';
|
||||
import { pickModel } from './models.js';
|
||||
import { runPipeline, realDeps } from './pipeline.js';
|
||||
|
||||
async function main(): Promise<void> {
|
||||
const runId = process.argv[2];
|
||||
if (!runId) {
|
||||
// 没 runId 没法定位 job.json / outbox，无处汇报——只能直接退出（非 0 让父进程可感知）。
|
||||
process.stderr.write('worker: 缺少 runId 参数（用法：worker <runId>）\n');
|
||||
process.exit(2);
|
||||
}
|
||||
|
||||
const emit = (p: OutboxPayload): void => { appendOutbox(runId, p); };
|
||||
const job = readJobSpec(runId);
|
||||
|
||||
// 起手汇报：pid / worktree / branch / 实际执行模型
|
||||
emit({
|
||||
type: 'started',
|
||||
pid: process.pid,
|
||||
worktree: job.worktreeDir,
|
||||
branch: job.branch,
|
||||
model: pickModel(job.task, job.project, 'executor'),
|
||||
});
|
||||
|
||||
// 心跳：立即一次 + 周期刷（结束时清掉）
|
||||
touchHeartbeat(runId);
|
||||
const hb = setInterval(() => touchHeartbeat(runId), HEARTBEAT_INTERVAL_MS);
|
||||
// 心跳定时器不应拖住事件循环退出（正常路径我们显式 exit，这里只是兜底）
|
||||
hb.unref?.();
|
||||
|
||||
// SIGTERM＝daemon 取消/超时：兜底报失败终态后干净退出（exit 0：已自报终态，不算崩溃）
|
||||
let signalled = false;
|
||||
process.on('SIGTERM', () => {
|
||||
if (signalled) return;
|
||||
signalled = true;
|
||||
clearInterval(hb);
|
||||
try {
|
||||
emit({ type: 'failed', error: 'worker 收到 SIGTERM（取消/超时）', transcriptRef: null, sessionId: null });
|
||||
emit({ type: 'done' });
|
||||
} catch { /* 汇报失败也要退出，别卡死 */ }
|
||||
process.exit(0);
|
||||
});
|
||||
|
||||
try {
|
||||
// runPipeline 内部已 emit failed|result + done；正常路径这里不再补发终态
|
||||
await runPipeline(job, realDeps, emit);
|
||||
} catch (e) {
|
||||
// 意外异常（pipeline 之外或 emit 抛错等）：兜底补一条 failed + done
|
||||
if (!signalled) {
|
||||
try {
|
||||
emit({ type: 'failed', error: (e as Error).message, transcriptRef: null, sessionId: null });
|
||||
emit({ type: 'done' });
|
||||
} catch { /* 已尽力汇报 */ }
|
||||
}
|
||||
} finally {
|
||||
clearInterval(hb);
|
||||
}
|
||||
|
||||
if (!signalled) process.exit(0);
|
||||
}
|
||||
|
||||
main().catch((e) => {
|
||||
// main 自身（如 readJobSpec 抛错）兜底：尽量写一条 failed，但若连 runId 都没有就只能干退。
|
||||
const runId = process.argv[2];
|
||||
if (runId) {
|
||||
try {
|
||||
appendOutbox(runId, { type: 'failed', error: (e as Error).message, transcriptRef: null, sessionId: null });
|
||||
appendOutbox(runId, { type: 'done' });
|
||||
} catch { /* 无处可报 */ }
|
||||
} else {
|
||||
process.stderr.write(`worker 致命错误：${(e as Error).message}\n`);
|
||||
}
|
||||
process.exit(1);
|
||||
});
|
||||
@@ -70,6 +70,7 @@ export interface Task {
|
||||
result: TaskResult | null;
|
||||
assignee: 'agent' | 'human' | null;
|
||||
retryBaseline: number; // 上次手动重投时已有的失败 run 数（重置重试计数用）
|
||||
nextEligibleAt: string | null; // 持久化退避：早于此时间不被领取（重试退避，重启不丢）；null=即刻可领
|
||||
createdAt: string;
|
||||
updatedAt: string;
|
||||
}
|
||||
@@ -89,6 +90,8 @@ export interface Run {
|
||||
transcriptRef: string | null; // agent 转录日志文件路径
|
||||
claudeSessionId: string | null;
|
||||
error: string | null;
|
||||
workerPid: number | null; // 执行该 run 的 worker 进程 pid（多进程执行；daemon 写，判活用）
|
||||
lastSeq: number; // 已 ingest 的 outbox 最大 seq（daemon 写，幂等游标；重启续读）
|
||||
}
|
||||
|
||||
export type EventType =
|
||||
|
||||
@@ -29,6 +29,9 @@ export function openDb(file: string): Database.Database {
|
||||
ensureColumn(db, 'projects', 'max_retries', 'max_retries INTEGER NOT NULL DEFAULT 2'); // 失败后最大自动重试次数
|
||||
ensureColumn(db, 'projects', 'timeout_ms', 'timeout_ms INTEGER NOT NULL DEFAULT 1800000'); // 单次执行超时毫秒
|
||||
ensureColumn(db, 'tasks', 'retry_baseline', 'retry_baseline INTEGER NOT NULL DEFAULT 0'); // 手动重投时的失败基线
|
||||
ensureColumn(db, 'tasks', 'next_eligible_at', 'next_eligible_at TEXT'); // 持久化退避（多进程执行）
|
||||
ensureColumn(db, 'runs', 'worker_pid', 'worker_pid INTEGER'); // worker 进程 pid
|
||||
ensureColumn(db, 'runs', 'last_seq', 'last_seq INTEGER NOT NULL DEFAULT 0'); // outbox ingest 游标
|
||||
const schema = readFileSync(join(HERE, 'schema.sql'), 'utf8');
|
||||
db.exec(schema);
|
||||
return db;
|
||||
|
||||
@@ -16,6 +16,7 @@ export interface TaskRow {
|
||||
title: string; complexity: string; status: string; priority: number;
|
||||
deps: string; plan: string | null; spec: string | null; operations: string | null;
|
||||
result: string | null; assignee: string | null; retry_baseline: number;
|
||||
next_eligible_at: string | null;
|
||||
created_at: string; updated_at: string;
|
||||
source_ref: string | null;
|
||||
}
|
||||
@@ -27,6 +28,7 @@ export interface RunRow {
|
||||
id: string; task_id: string; kind: string; worktree: string | null; branch: string | null;
|
||||
status: string; started_at: string; ended_at: string | null;
|
||||
transcript_ref: string | null; claude_session_id: string | null; error: string | null;
|
||||
worker_pid: number | null; last_seq: number;
|
||||
}
|
||||
export interface EventRow {
|
||||
id: string; project_id: string; task_id: string | null; type: string; payload: string; at: string;
|
||||
@@ -68,6 +70,7 @@ export function rowToTask(r: TaskRow, approvals: ApprovalRecord[] = []): Task {
|
||||
plan: r.plan, spec: r.spec, operations: r.operations,
|
||||
approvals, result: r.result ? parseResult(r.result) : null,
|
||||
assignee: r.assignee as Task['assignee'], retryBaseline: r.retry_baseline ?? 0,
|
||||
nextEligibleAt: r.next_eligible_at ?? null,
|
||||
createdAt: r.created_at, updatedAt: r.updated_at,
|
||||
};
|
||||
}
|
||||
@@ -85,6 +88,7 @@ export function rowToRun(r: RunRow): Run {
|
||||
worktree: r.worktree, branch: r.branch, status: r.status as RunStatus,
|
||||
startedAt: r.started_at, endedAt: r.ended_at,
|
||||
transcriptRef: r.transcript_ref, claudeSessionId: r.claude_session_id, error: r.error,
|
||||
workerPid: r.worker_pid ?? null, lastSeq: r.last_seq ?? 0,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -36,6 +36,7 @@ CREATE TABLE IF NOT EXISTS tasks (
|
||||
result TEXT, -- JSON TaskResult
|
||||
assignee TEXT, -- agent | human
|
||||
retry_baseline INTEGER NOT NULL DEFAULT 0, -- 上次手动重投时已有的失败 run 数（重置重试计数用）
|
||||
next_eligible_at TEXT, -- 持久化退避：早于此时间不被领取（null=即刻可领）
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL,
|
||||
source_ref TEXT -- 旧 todo 来源标识（todo:17 / todo:17/1A），项目内唯一
|
||||
@@ -66,7 +67,9 @@ CREATE TABLE IF NOT EXISTS runs (
|
||||
ended_at TEXT,
|
||||
transcript_ref TEXT,
|
||||
claude_session_id TEXT,
|
||||
error TEXT
|
||||
error TEXT,
|
||||
worker_pid INTEGER, -- 多进程执行：worker 进程 pid（daemon 写，判活用）
|
||||
last_seq INTEGER NOT NULL DEFAULT 0 -- 已 ingest 的 outbox 最大 seq（幂等游标）
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_runs_task ON runs(task_id, started_at);
|
||||
|
||||
|
||||
+92
-10
@@ -231,8 +231,8 @@ export class Store {
|
||||
id: id('tsk'), project_id: input.projectId, parent_id: input.parentId ?? null, depth,
|
||||
title: input.title, complexity: input.complexity, status, priority: input.priority ?? 1,
|
||||
deps: JSON.stringify(input.deps ?? []), plan: null, spec: null, operations: null,
|
||||
result: null, assignee: null, retry_baseline: 0, created_at: now(), updated_at: now(),
|
||||
source_ref: null,
|
||||
result: null, assignee: null, retry_baseline: 0, next_eligible_at: null,
|
||||
created_at: now(), updated_at: now(), source_ref: null,
|
||||
};
|
||||
this.db.prepare(
|
||||
`INSERT INTO tasks (id,project_id,parent_id,depth,title,complexity,status,priority,deps,plan,spec,operations,result,assignee,retry_baseline,created_at,updated_at,source_ref)
|
||||
@@ -562,15 +562,86 @@ export class Store {
|
||||
* 卡在 executing 的任务转 failed→queued 等编排器重新领取。
|
||||
* 注：中断产生的 failed run 会计入该任务的失败次数（多次中断+真失败可能提前转 needs_attention，可接受）。
|
||||
*/
|
||||
reconcileInterrupted(): { runs: number; tasks: number } {
|
||||
const runs = this.db.prepare(`SELECT * FROM runs WHERE status = 'started'`).all() as RunRow[];
|
||||
for (const r of runs) this.finishRun(r.id, 'failed', { error: 'daemon 重启，执行中断' });
|
||||
const rows = this.db.prepare(`SELECT * FROM tasks WHERE status = 'executing'`).all() as TaskRow[];
|
||||
for (const t of rows) {
|
||||
this.transition(t.id, 'failed', { auto: 'daemon-restart' });
|
||||
this.transition(t.id, 'queued', { auto: 'daemon-restart' });
|
||||
/**
|
||||
* 失败/重试策略的唯一落点（ingest 的 failed 记录、reaper 判死、reconcile 判死 都调它）。
|
||||
* 把（可选）执行 run 收尾为 failed，按「净失败次数 vs project.maxRetries」决定：
|
||||
* 未到上限 → queued + 持久化退避（next_eligible_at = now + 指数退避）
|
||||
* 到上限 → needs_attention（清退避）
|
||||
* netFailed 相对 retry_baseline（手动重投时设的基线）。退避：min(30s·2^(n-1), 10min)。
|
||||
*/
|
||||
failTaskAttempt(taskId: string, runId: string | null, error: string): Task {
|
||||
const row = this.getTaskRow(taskId);
|
||||
if (!row) throw new StoreError(`任务不存在: ${taskId}`);
|
||||
const project = this.getProject(row.project_id);
|
||||
if (!project) throw new StoreError(`项目不存在: ${row.project_id}`);
|
||||
|
||||
// 收尾执行 run：给了 runId 且仍 started → 标 failed；没给 → 补记一条 failed（保证重试计数不漏）
|
||||
if (runId) {
|
||||
const rr = this.db.prepare(`SELECT status FROM runs WHERE id = ?`).get(runId) as { status: string } | undefined;
|
||||
if (rr && rr.status === 'started') this.finishRun(runId, 'failed', { error });
|
||||
} else {
|
||||
const r = this.startRun(taskId, 'executor');
|
||||
this.finishRun(r.id, 'failed', { error });
|
||||
}
|
||||
return { runs: runs.length, tasks: rows.length };
|
||||
|
||||
if ((this.getTaskRow(taskId)!.status as TaskStatus) !== 'failed') {
|
||||
this.transition(taskId, 'failed', { by: 'failTaskAttempt', error });
|
||||
}
|
||||
|
||||
const allFailed = (this.db.prepare(
|
||||
`SELECT COUNT(*) AS n FROM runs WHERE task_id = ? AND kind = 'executor' AND status = 'failed'`,
|
||||
).get(taskId) as { n: number }).n;
|
||||
const netFailed = Math.max(0, allFailed - row.retry_baseline);
|
||||
const priorNetFailed = Math.max(0, netFailed - 1); // 不含本次
|
||||
if (priorNetFailed < project.maxRetries) {
|
||||
const attempt = priorNetFailed + 1;
|
||||
const backoffMs = Math.min(30_000 * 2 ** (attempt - 1), 10 * 60_000);
|
||||
this.setNextEligibleAt(taskId, new Date(Date.now() + backoffMs).toISOString());
|
||||
this.transition(taskId, 'queued', { by: 'failTaskAttempt', retry: attempt, backoffMs });
|
||||
} else {
|
||||
this.setNextEligibleAt(taskId, null);
|
||||
this.transition(taskId, 'needs_attention', { by: 'failTaskAttempt', allFailed, netFailed });
|
||||
}
|
||||
return this.getTask(taskId)!;
|
||||
}
|
||||
|
||||
/** 持久化退避：早于 next_eligible_at 不被领取（null=即刻可领）。 */
|
||||
setNextEligibleAt(taskId: string, iso: string | null): void {
|
||||
this.db.prepare(`UPDATE tasks SET next_eligible_at = ?, updated_at = ? WHERE id = ?`).run(iso, now(), taskId);
|
||||
}
|
||||
|
||||
/** 项目内 executing 任务数（多进程执行的并发闸：每个 executing 对应一个活 worker）。 */
|
||||
countExecuting(projectId: string): number {
|
||||
return (this.db.prepare(
|
||||
`SELECT COUNT(*) AS n FROM tasks WHERE project_id = ? AND status = 'executing'`,
|
||||
).get(projectId) as { n: number }).n;
|
||||
}
|
||||
|
||||
/** 所有 executing 任务 + 其最近一条 executor run（reaper / reconcile 判活用）。run 可能为 null（异常）。 */
|
||||
executingWithLatestExecutorRun(): Array<{ task: Task; run: Run | null }> {
|
||||
const rows = this.db.prepare(`SELECT * FROM tasks WHERE status = 'executing'`).all() as TaskRow[];
|
||||
return rows.map((t) => {
|
||||
const rr = this.db.prepare(
|
||||
`SELECT * FROM runs WHERE task_id = ? AND kind = 'executor' ORDER BY started_at DESC LIMIT 1`,
|
||||
).get(t.id) as RunRow | undefined;
|
||||
return { task: rowToTask(t), run: rr ? rowToRun(rr) : null };
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 启动中断恢复（多进程执行）：逐个 executing 任务按注入的 isAlive 判活——
|
||||
* 活 → re-adopt（保留 executing，daemon 续 ingest 其 outbox）；
|
||||
* 死 → failTaskAttempt（收尾 + 重试/needs_attention）。
|
||||
* isAlive 由 daemon 注入（protocol.isWorkerAlive，基于 worker_pid + 心跳）；默认保守判死（无判活信息时回收）。
|
||||
*/
|
||||
reconcileInterrupted(isAlive: (run: Run | null) => boolean = () => false): { readopted: number; reclaimed: number } {
|
||||
let readopted = 0, reclaimed = 0;
|
||||
for (const { task, run } of this.executingWithLatestExecutorRun()) {
|
||||
if (run && isAlive(run)) { readopted++; continue; }
|
||||
this.failTaskAttempt(task.id, run?.id ?? null, 'daemon 重启时发现 worker 已退出');
|
||||
reclaimed++;
|
||||
}
|
||||
return { readopted, reclaimed };
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -687,6 +758,7 @@ export class Store {
|
||||
const rr: RunRow = {
|
||||
id: id('run'), task_id: taskId, kind, worktree: fields.worktree ?? null, branch: fields.branch ?? null,
|
||||
status: 'started', started_at: now(), ended_at: null, transcript_ref: null, claude_session_id: null, error: null,
|
||||
worker_pid: null, last_seq: 0,
|
||||
};
|
||||
this.db.prepare(
|
||||
`INSERT INTO runs (id,task_id,kind,worktree,branch,status,started_at,ended_at,transcript_ref,claude_session_id,error)
|
||||
@@ -707,6 +779,16 @@ export class Store {
|
||||
return rowToRun(this.db.prepare(`SELECT * FROM runs WHERE id = ?`).get(runId) as RunRow);
|
||||
}
|
||||
|
||||
/** 记录执行该 run 的 worker 进程 pid（多进程执行，daemon spawn 后写；判活用）。 */
|
||||
setWorkerPid(runId: string, pid: number): void {
|
||||
this.db.prepare(`UPDATE runs SET worker_pid = ? WHERE id = ?`).run(pid, runId);
|
||||
}
|
||||
|
||||
/** 更新已 ingest 的 outbox 最大 seq（幂等游标；daemon ingest 后写，重启续读）。 */
|
||||
setLastSeq(runId: string, seq: number): void {
|
||||
this.db.prepare(`UPDATE runs SET last_seq = ? WHERE id = ?`).run(seq, runId);
|
||||
}
|
||||
|
||||
/** 所有进行中的 run（status='started'），联 tasks 取任务标题与项目。 */
|
||||
activeRuns(): ActiveRun[] {
|
||||
const rows = this.db.prepare(
|
||||
|
||||
Vendored
+37
@@ -0,0 +1,37 @@
|
||||
// 集成测试用的「假 worker」：走完整 outbox 协议，但【不调用 SDK / 不建 worktree】，零 token。
|
||||
// 用法：daemon 通过 MAESTRO_WORKER_CMD='npx tsx test/fixtures/fake-worker.ts' spawn 它，argv[2]=runId。
|
||||
// 行为：emit started + 持续刷心跳；轮询 runs/<runId>/finish 哨兵文件出现后 emit result+done 退出。
|
||||
// 收到 SIGTERM → emit failed+done 退出。模拟一个「在跑的真进程」，供测试验证 spawn / 心跳 / re-adopt / ingest。
|
||||
import { existsSync } from 'node:fs';
|
||||
import { join } from 'node:path';
|
||||
import { appendOutbox, readJobSpec, touchHeartbeat, runDir } from '../../src/executor/protocol.js';
|
||||
|
||||
const runId = process.argv[2];
|
||||
if (!runId) { process.exit(2); }
|
||||
|
||||
const job = readJobSpec(runId);
|
||||
touchHeartbeat(runId);
|
||||
appendOutbox(runId, { type: 'started', pid: process.pid, worktree: job.worktreeDir, branch: job.branch, model: 'fake' });
|
||||
|
||||
const hb = setInterval(() => touchHeartbeat(runId), 300);
|
||||
const finishFile = join(runDir(runId), 'finish');
|
||||
const poll = setInterval(() => {
|
||||
if (!existsSync(finishFile)) return;
|
||||
clearInterval(hb); clearInterval(poll);
|
||||
appendOutbox(runId, {
|
||||
type: 'result', branch: job.branch, worktree: job.worktreeDir,
|
||||
diffSummary: ' demo.txt | 1 +', commits: ['abc1234 fake commit'],
|
||||
executor: { transcriptRef: null, sessionId: 'sess-fake' },
|
||||
code: { summary: 'code ok', verdict: 'approve', transcriptRef: null },
|
||||
security: { summary: 'sec ok', verdict: 'approve', transcriptRef: null },
|
||||
});
|
||||
appendOutbox(runId, { type: 'done' });
|
||||
process.exit(0);
|
||||
}, 80);
|
||||
|
||||
process.on('SIGTERM', () => {
|
||||
clearInterval(hb); clearInterval(poll);
|
||||
appendOutbox(runId, { type: 'failed', error: 'worker 收到 SIGTERM', transcriptRef: null, sessionId: null });
|
||||
appendOutbox(runId, { type: 'done' });
|
||||
process.exit(0);
|
||||
});
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
+292
-398
@@ -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<T>(): { promise: Promise<T>; resolve: (v: T) => void } {
|
||||
let resolve!: (v: T) => void;
|
||||
const promise = new Promise<T>((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> = {}): 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<OrchestratorDeps> = {}): { 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<void> {
|
||||
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<void>();
|
||||
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();
|
||||
});
|
||||
|
||||
@@ -0,0 +1,183 @@
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { runPipeline, type PipelineDeps } from '../src/executor/pipeline.js';
|
||||
import type { JobSpec, OutboxPayload } from '../src/executor/protocol.js';
|
||||
import type { Project, Task } from '../src/model/types.js';
|
||||
import type { RunnerResult } from '../src/executor/runner.js';
|
||||
import type { ReviewResult } from '../src/executor/reviewer.js';
|
||||
|
||||
// ───────────────────────── 固定 fixtures ─────────────────────────
|
||||
|
||||
const TASK_ID = 'task-pipe-1';
|
||||
const RUN_ID = 'run-pipe-1';
|
||||
const FAKE_DIR = `/tmp/fake-wt/${TASK_ID}`;
|
||||
const FAKE_BRANCH = `maestro/${TASK_ID}`;
|
||||
|
||||
const fakeProject: Project = {
|
||||
id: 'proj-1', name: 'pipe', repoPath: '/tmp/pipe-repo', defaultBranch: 'main',
|
||||
verifyCmd: null, autonomy: 'auto-easy', model: null, concurrency: 1, maxRetries: 2,
|
||||
timeoutMs: 1_800_000, status: 'active', logo: null, sortOrder: 0,
|
||||
createdAt: '2024-01-01T00:00:00.000Z', lastSyncAt: null,
|
||||
};
|
||||
|
||||
const fakeTask: Task = {
|
||||
id: TASK_ID, projectId: 'proj-1', parentId: null, depth: 1, title: 'tweak',
|
||||
complexity: 'easy', status: 'queued', priority: 0, deps: [],
|
||||
plan: null, spec: null, operations: '在 README.md 追加一行', approvals: [],
|
||||
result: null, assignee: 'agent', retryBaseline: 0, nextEligibleAt: null,
|
||||
createdAt: '2024-01-01T00:00:00.000Z', updatedAt: '2024-01-01T00:00:00.000Z',
|
||||
};
|
||||
|
||||
const fakeJob: JobSpec = {
|
||||
runId: RUN_ID, task: fakeTask, project: fakeProject,
|
||||
worktreeDir: FAKE_DIR, branch: FAKE_BRANCH,
|
||||
};
|
||||
|
||||
const okRun: RunnerResult = {
|
||||
ok: true, transcriptRef: '/tmp/fake.jsonl', sessionId: 'sess-exec-1',
|
||||
finalText: '执行自述：改了 README',
|
||||
};
|
||||
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<PipelineDeps> = {}): PipelineDeps {
|
||||
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 }),
|
||||
runTask: async () => okRun,
|
||||
reviewCode: async () => okReview,
|
||||
reviewSecurity: async () => okSecurity,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
/** 收集 emit 出的 payload 序列；返回收集器与 emit 函数。 */
|
||||
function collector(): { emitted: OutboxPayload[]; emit: (p: OutboxPayload) => void } {
|
||||
const emitted: OutboxPayload[] = [];
|
||||
return { emitted, emit: (p) => { emitted.push(p); } };
|
||||
}
|
||||
|
||||
/** 取某类型的（首条）payload。 */
|
||||
function find<T extends OutboxPayload['type']>(
|
||||
emitted: OutboxPayload[], type: T,
|
||||
): Extract<OutboxPayload, { type: T }> | undefined {
|
||||
return emitted.find((p) => p.type === type) as Extract<OutboxPayload, { type: T }> | undefined;
|
||||
}
|
||||
|
||||
// ───────────────────────── 用例 ─────────────────────────
|
||||
|
||||
test('成功路径：emit result（四要素正确）+ 末尾 done，不含 failed', async () => {
|
||||
const { emitted, emit } = collector();
|
||||
await runPipeline(fakeJob, mockDeps(), emit);
|
||||
|
||||
// 不含 failed
|
||||
assert.equal(find(emitted, 'failed'), undefined, '成功路径不应 emit failed');
|
||||
|
||||
// 末尾是 done
|
||||
assert.equal(emitted.at(-1)?.type, 'done', '最后一条应为 done');
|
||||
|
||||
// result 四要素
|
||||
const result = find(emitted, 'result');
|
||||
assert.ok(result, '应 emit result');
|
||||
assert.equal(result.branch, FAKE_BRANCH);
|
||||
assert.equal(result.worktree, FAKE_DIR);
|
||||
assert.equal(result.diffSummary, ' README.md | 1 +');
|
||||
assert.deepEqual(result.commits, ['abc1234 hello maestro']);
|
||||
// executor 透传 runner 的 transcriptRef/sessionId
|
||||
assert.deepEqual(result.executor, { transcriptRef: '/tmp/fake.jsonl', sessionId: 'sess-exec-1' });
|
||||
// code review report（ReviewResult → ReviewReport：丢 sessionId）
|
||||
assert.deepEqual(result.code, {
|
||||
summary: '## 做了什么\nmock 复审通过', verdict: 'approve', transcriptRef: '/tmp/fake-review.jsonl',
|
||||
});
|
||||
// security report
|
||||
assert.deepEqual(result.security, {
|
||||
summary: '## 安全审计\nmock 审计通过', verdict: 'approve', transcriptRef: '/tmp/fake-security.jsonl',
|
||||
});
|
||||
|
||||
// done 恰好一条
|
||||
assert.equal(emitted.filter((p) => p.type === 'done').length, 1);
|
||||
});
|
||||
|
||||
test('双复审收到的执行者自述＝runner.finalText', async () => {
|
||||
let codeReport: string | null = null;
|
||||
let secReport: string | null = null;
|
||||
const { emit } = collector();
|
||||
await runPipeline(fakeJob, mockDeps({
|
||||
reviewCode: async (_t, _p, _wt, _runId, report) => { codeReport = report; return okReview; },
|
||||
reviewSecurity: async (_t, _p, _wt, _runId, report) => { secReport = report; return okSecurity; },
|
||||
}), emit);
|
||||
assert.equal(codeReport, '执行自述：改了 README');
|
||||
assert.equal(secReport, '执行自述：改了 README');
|
||||
});
|
||||
|
||||
test('executor 失败：emit failed（携带 error/transcriptRef/sessionId）+ done，无 result', async () => {
|
||||
const { emitted, emit } = collector();
|
||||
await runPipeline(fakeJob, mockDeps({
|
||||
runTask: async () => ({ ok: false, transcriptRef: '/tmp/exec-fail.jsonl', sessionId: 'sess-x', error: 'boom 执行炸了' }),
|
||||
}), emit);
|
||||
|
||||
assert.equal(find(emitted, 'result'), undefined, 'executor 失败不应 emit result');
|
||||
const failed = find(emitted, 'failed');
|
||||
assert.ok(failed, '应 emit failed');
|
||||
assert.equal(failed.error, 'boom 执行炸了');
|
||||
assert.equal(failed.transcriptRef, '/tmp/exec-fail.jsonl');
|
||||
assert.equal(failed.sessionId, 'sess-x');
|
||||
assert.equal(emitted.at(-1)?.type, 'done', '末尾应为 done');
|
||||
});
|
||||
|
||||
test('verify 失败：runner ok 但 verify !ok → emit failed（沿用 executor transcript）+ done，无 result', async () => {
|
||||
const { emitted, emit } = collector();
|
||||
await runPipeline(fakeJob, mockDeps({
|
||||
verify: async () => ({ ok: false, exitCode: 1, logRef: '/tmp/v.log', error: 'verify 失败（exit 1）' }),
|
||||
}), emit);
|
||||
|
||||
assert.equal(find(emitted, 'result'), undefined, 'verify 失败不应 emit result');
|
||||
const failed = find(emitted, 'failed');
|
||||
assert.ok(failed, '应 emit failed');
|
||||
assert.equal(failed.error, 'verify 失败（exit 1）');
|
||||
// transcriptRef/sessionId 沿用 executor 的（okRun）
|
||||
assert.equal(failed.transcriptRef, '/tmp/fake.jsonl');
|
||||
assert.equal(failed.sessionId, 'sess-exec-1');
|
||||
assert.equal(emitted.at(-1)?.type, 'done', '末尾应为 done');
|
||||
});
|
||||
|
||||
test('复审抛错不挡：reviewCode 抛错 → result 仍 emit，code.verdict=null 且 summary 含「自动复审失败」', async () => {
|
||||
const { emitted, emit } = collector();
|
||||
await runPipeline(fakeJob, mockDeps({
|
||||
reviewCode: async () => { throw new Error('复审 CC 崩了'); },
|
||||
}), emit);
|
||||
|
||||
const result = find(emitted, 'result');
|
||||
assert.ok(result, '复审抛错仍应 emit result');
|
||||
assert.equal(result.code.verdict, null);
|
||||
assert.equal(result.code.transcriptRef, null);
|
||||
assert.match(result.code.summary ?? '', /自动复审失败/);
|
||||
assert.match(result.code.summary ?? '', /复审 CC 崩了/);
|
||||
// 另一个复审不受影响
|
||||
assert.equal(result.security.verdict, 'approve');
|
||||
assert.equal(result.security.summary, '## 安全审计\nmock 审计通过');
|
||||
assert.equal(find(emitted, 'failed'), undefined, '复审失败不应升级为 failed');
|
||||
assert.equal(emitted.at(-1)?.type, 'done');
|
||||
});
|
||||
|
||||
test('安全审计抛错不挡：security.verdict=null、summary 含「自动复审失败」，code review 不受影响', async () => {
|
||||
const { emitted, emit } = collector();
|
||||
await runPipeline(fakeJob, mockDeps({
|
||||
reviewSecurity: async () => { throw new Error('审计 CC 崩了'); },
|
||||
}), emit);
|
||||
|
||||
const result = find(emitted, 'result');
|
||||
assert.ok(result);
|
||||
assert.equal(result.security.verdict, null);
|
||||
assert.match(result.security.summary ?? '', /自动复审失败：审计 CC 崩了/);
|
||||
assert.equal(result.code.verdict, 'approve');
|
||||
assert.equal(emitted.at(-1)?.type, 'done');
|
||||
});
|
||||
@@ -0,0 +1,140 @@
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { execFileSync } from 'node:child_process';
|
||||
import { existsSync, mkdtempSync, rmSync, writeFileSync, utimesSync } from 'node:fs';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join, resolve } from 'node:path';
|
||||
import { Store } from '../src/store/index.js';
|
||||
import { createOrchestrator } from '../src/daemon/orchestrator.js';
|
||||
import { ingestAll } from '../src/daemon/ingest.js';
|
||||
import {
|
||||
readOutboxAll, heartbeatAgeMs, isWorkerAlive, runDir, heartbeatPath, pidAlive,
|
||||
} from '../src/executor/protocol.js';
|
||||
import type { Run } from '../src/model/types.js';
|
||||
|
||||
const log = { info: (): void => undefined, error: (): void => undefined };
|
||||
const FIXTURE = resolve(process.cwd(), 'test/fixtures/fake-worker.ts');
|
||||
|
||||
/** 真判活：与 daemon index.ts 注入 reconcileInterrupted 的一致 */
|
||||
function realIsAlive(run: Run | null): boolean {
|
||||
return run !== null && isWorkerAlive({
|
||||
pid: run.workerPid,
|
||||
heartbeatAgeMs: heartbeatAgeMs(run.id),
|
||||
startedAgeMs: Date.now() - Date.parse(run.startedAt),
|
||||
});
|
||||
}
|
||||
|
||||
async function waitFor(pred: () => boolean, timeoutMs = 20_000, stepMs = 150): Promise<boolean> {
|
||||
const end = Date.now() + timeoutMs;
|
||||
while (Date.now() < end) {
|
||||
if (pred()) return true;
|
||||
await new Promise((r) => setTimeout(r, stepMs));
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
function makeRepo(dir: string): void {
|
||||
const g = (args: string[]): void => { execFileSync('git', args, { cwd: dir }); };
|
||||
g(['init', '-b', 'main']); g(['config', 'user.name', 't']); g(['config', 'user.email', 't@t']);
|
||||
writeFileSync(join(dir, 'README.md'), '# demo\n');
|
||||
g(['add', '-A']); g(['commit', '-m', 'init']);
|
||||
}
|
||||
|
||||
/** 公共 setup：临时 data dir + repo + MAESTRO_WORKER_CMD 指向假 worker；返回 store/task/cleanup */
|
||||
function setupReal(t: { after: (fn: () => void) => void }): { store: Store; dbFile: string; taskId: string; repo: string } {
|
||||
const dataDir = mkdtempSync(join(tmpdir(), 'maestro-wi-data-'));
|
||||
const repo = mkdtempSync(join(tmpdir(), 'maestro-wi-repo-'));
|
||||
makeRepo(repo);
|
||||
const prevData = process.env.MAESTRO_DATA_DIR;
|
||||
const prevCmd = process.env.MAESTRO_WORKER_CMD;
|
||||
process.env.MAESTRO_DATA_DIR = dataDir;
|
||||
process.env.MAESTRO_WORKER_CMD = `npx tsx ${FIXTURE}`;
|
||||
const dbFile = join(dataDir, 'maestro.sqlite');
|
||||
const store = new Store(dbFile);
|
||||
const p = store.createProject({ name: 'wi', repoPath: repo, autonomy: 'auto-approved', concurrency: 1 });
|
||||
const task = store.createTask({ projectId: p.id, title: '集成任务', complexity: 'easy' });
|
||||
store.setOperations(task.id, '改一行'); // → ready
|
||||
|
||||
t.after(() => {
|
||||
// 收尾：杀掉可能残留的 worker
|
||||
try {
|
||||
const run = store.listRuns(task.id).find((r) => r.kind === 'executor');
|
||||
if (run?.workerPid && pidAlive(run.workerPid)) process.kill(run.workerPid, 'SIGKILL');
|
||||
} catch { /* noop */ }
|
||||
try { store.close(); } catch { /* noop */ }
|
||||
if (prevData === undefined) delete process.env.MAESTRO_DATA_DIR; else process.env.MAESTRO_DATA_DIR = prevData;
|
||||
if (prevCmd === undefined) delete process.env.MAESTRO_WORKER_CMD; else process.env.MAESTRO_WORKER_CMD = prevCmd;
|
||||
rmSync(dataDir, { recursive: true, force: true });
|
||||
rmSync(repo, { recursive: true, force: true });
|
||||
});
|
||||
return { store, dbFile, taskId: task.id, repo };
|
||||
}
|
||||
|
||||
test('集成：spawn 真 worker 进程 → daemon「重启」中 re-adopt 存活 worker → finish → ingest 到 exec_review', async (t) => {
|
||||
const { store, dbFile, taskId } = setupReal(t);
|
||||
|
||||
// 1) 监工领取 → 真 spawn 假 worker 进程（经 MAESTRO_WORKER_CMD）
|
||||
createOrchestrator(store, log).claimTick();
|
||||
const run = store.listRuns(taskId).find((r) => r.kind === 'executor');
|
||||
assert.ok(run, '应建了 executor run');
|
||||
assert.equal(store.getTask(taskId)!.status, 'executing');
|
||||
assert.ok(run!.workerPid, 'setWorkerPid 已写 worker 进程 pid');
|
||||
|
||||
// 2) 等 worker 进程启动并 emit started（npx tsx 冷启动较慢）
|
||||
const started = await waitFor(() => readOutboxAll(run!.id).some((r) => r.type === 'started'));
|
||||
assert.ok(started, 'worker 应在超时内 emit started');
|
||||
assert.ok(pidAlive(run!.workerPid!), 'worker 进程应存活');
|
||||
|
||||
// 3) 模拟 daemon 重启：换一个 Store 连同一个 db，reconcile 用真判活
|
||||
store.close();
|
||||
const store2 = new Store(dbFile);
|
||||
t.after(() => { try { store2.close(); } catch { /* noop */ } });
|
||||
const rec = store2.reconcileInterrupted(realIsAlive);
|
||||
assert.equal(rec.readopted, 1, '存活 worker 应被 re-adopt');
|
||||
assert.equal(rec.reclaimed, 0, '不应回收存活 worker');
|
||||
assert.equal(store2.getTask(taskId)!.status, 'executing', 're-adopt 后任务仍在 executing（未被打断/重跑）');
|
||||
|
||||
// 4) 投放 finish 哨兵 → worker emit result + done + 退出
|
||||
writeFileSync(join(runDir(run!.id), 'finish'), '');
|
||||
const got = await waitFor(() => readOutboxAll(run!.id).some((r) => r.type === 'result'));
|
||||
assert.ok(got, 'worker 应在 finish 后 emit result');
|
||||
|
||||
// 5) daemon ingest → 任务进 exec_review，结果四字段就位
|
||||
ingestAll(store2, log);
|
||||
const done = store2.getTask(taskId)!;
|
||||
assert.equal(done.status, 'exec_review');
|
||||
assert.equal(done.result?.verdict, 'approve');
|
||||
assert.equal(done.result?.securityVerdict, 'approve');
|
||||
assert.match(done.result?.diffSummary ?? '', /demo\.txt/);
|
||||
// reviewer + security run 各一条 succeeded（daemon 据 result 建）
|
||||
const runs = store2.listRuns(taskId);
|
||||
assert.ok(runs.some((r) => r.kind === 'reviewer' && r.status === 'succeeded'));
|
||||
assert.ok(runs.some((r) => r.kind === 'security' && r.status === 'succeeded'));
|
||||
assert.ok(runs.some((r) => r.kind === 'executor' && r.status === 'succeeded'));
|
||||
});
|
||||
|
||||
test('集成：worker 进程死亡（心跳变陈旧）→ reconcile 回收 → 任务退避重入队', async (t) => {
|
||||
const { store, dbFile, taskId } = setupReal(t);
|
||||
createOrchestrator(store, log).claimTick();
|
||||
const run = store.listRuns(taskId).find((r) => r.kind === 'executor')!;
|
||||
await waitFor(() => readOutboxAll(run.id).some((r) => r.type === 'started'));
|
||||
|
||||
// 杀掉整个 worker 进程组（detached spawn 使子进程为组长；测试经 `npx tsx` 是进程树，
|
||||
// 须杀组 -pid 才能连同孙进程一起灭，否则孙进程继续刷心跳。生产是 `node worker.js`，pid 即真 worker）。
|
||||
if (run.workerPid) { try { process.kill(-run.workerPid, 'SIGKILL'); } catch { /* 退回单 pid */ try { process.kill(run.workerPid, 'SIGKILL'); } catch { /* 已退出 */ } } }
|
||||
// 等进程真正退出、停止刷心跳后，再把心跳 mtime 调到 2 分钟前（越过 60s grace，模拟「死了且心跳陈旧」）
|
||||
await new Promise((r) => setTimeout(r, 500));
|
||||
const old = new Date(Date.now() - 120_000);
|
||||
if (existsSync(heartbeatPath(run.id))) utimesSync(heartbeatPath(run.id), old, old);
|
||||
|
||||
store.close();
|
||||
const store2 = new Store(dbFile);
|
||||
t.after(() => { try { store2.close(); } catch { /* noop */ } });
|
||||
const rec = store2.reconcileInterrupted(realIsAlive);
|
||||
assert.equal(rec.reclaimed, 1, '死 worker 应被回收');
|
||||
assert.equal(rec.readopted, 0);
|
||||
// 默认 maxRetries=2：第一次失败 → 退避重入队（queued + nextEligibleAt 有值）
|
||||
const tk = store2.getTask(taskId)!;
|
||||
assert.ok(['queued', 'blocked'].includes(tk.status), `回收后应重入队，实际 ${tk.status}`);
|
||||
assert.ok(tk.nextEligibleAt, '应写了持久化退避 nextEligibleAt');
|
||||
});
|
||||
Reference in New Issue
Block a user