import { test } from 'node:test'; import assert from 'node:assert/strict'; import { Store } from '../src/store/index.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 makeClock(initialMs = Date.now()): { nowMs: () => number; advance: (ms: number) => void } { let t = initialMs; return { nowMs: () => t, advance: (ms) => { t += ms; } }; } interface MockState { spawned: string[]; // 被 spawn 的 runId 列表 jobs: JobSpec[]; // 被 writeJobSpec 的 job 列表 ingestCalls: number; alive: boolean; // isWorkerAlive 的返回(reaper 用) nextPid: number; } /** 全 mock 依赖:不真 spawn / 不真判活 / ingest no-op / job 不落盘。可按用例覆盖。 */ function mockDeps(state: MockState, overrides: Partial = {}): { deps: OrchestratorDeps } { const deps: OrchestratorDeps = { spawnWorker: (runId: string): number => { state.spawned.push(runId); return state.nextPid++; }, isWorkerAlive: () => state.alive, ingestAll: () => { state.ingestCalls++; }, writeJobSpec: (job: JobSpec) => { state.jobs.push(job); }, nowMs: Date.now, ...overrides, }; return { deps }; } function freshState(): MockState { return { spawned: [], jobs: [], ingestCalls: 0, alive: true, nextPid: 1000 }; } function setup(autonomy: Autonomy, concurrency = 1): { store: Store; projectId: string } { const store = new Store(':memory:'); const p = store.createProject({ name: 'orch', repoPath: '/tmp/orch-repo-' + Math.random(), autonomy, concurrency, }); return { store, projectId: p.id }; } // ───────────────────────── workerEntry / workerEnv(可读测的小函数)───────────────────────── 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:透传会话/系统变量(keychain 可用),剥离密钥,保留鉴权变量', () => { const env = workerEnv({ PATH: '/usr/bin', HOME: '/home/u', LANG: 'en_US.UTF-8', USER: 'wj', LOGNAME: 'wj', __CF_USER_TEXT_ENCODING: '0x1F5:0:0', TMPDIR: '/tmp', XPC_SERVICE_NAME: '0', ANTHROPIC_API_KEY: 'sk-x', CLAUDE_CODE_OAUTH_TOKEN: 'oauth-x', CLAUDE_CODE_FOO: 'y', MAESTRO_ORCH_INTERVAL: '15', MAESTRO_DATA_DIR: '/data', SOME_SECRET: 'leak', AWS_SECRET_ACCESS_KEY: 'nope', GITHUB_TOKEN: 'gh', CF_API_TOKEN: 'cf', }); // 会话/系统变量透传(macOS keychain/Security 访问需要它们) for (const k of ['PATH', 'HOME', 'LANG', 'USER', 'LOGNAME', '__CF_USER_TEXT_ENCODING', 'TMPDIR', 'XPC_SERVICE_NAME']) { assert.ok(env[k] !== undefined, `会话变量 ${k} 应透传`); } // 鉴权变量始终保留(即便名含 TOKEN/KEY) assert.equal(env.ANTHROPIC_API_KEY, 'sk-x'); assert.equal(env.CLAUDE_CODE_OAUTH_TOKEN, 'oauth-x'); assert.equal(env.CLAUDE_CODE_FOO, 'y'); assert.equal(env.MAESTRO_DATA_DIR, '/data'); // worker 须看同一 runs/ 目录 assert.equal(env.MAESTRO_ORCH_INTERVAL, '15'); // 非密钥 → 透传(黑名单语义) // 真正的密钥类被剥离 assert.equal(env.SOME_SECRET, undefined, '*_SECRET 剥离'); assert.equal(env.AWS_SECRET_ACCESS_KEY, undefined, 'AWS 密钥剥离'); assert.equal(env.GITHUB_TOKEN, undefined, 'GITHUB_TOKEN 剥离'); assert.equal(env.CF_API_TOKEN, undefined, 'CF token 剥离'); }); // ───────────────────────── 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(); 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('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' }); store.setSpec(medium.id, '方案'); store.transition(medium.id, 'spec_review'); store.decide(medium.id, 'accept', 'user'); // medium → ready assert.equal(store.getTask(medium.id)!.status, 'ready'); const state = freshState(); const { deps } = mockDeps(state); createOrchestrator(store, noopLog, deps).claimTick(); 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('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'); const dep = store.createTask({ projectId, title: 'after-mid', complexity: 'easy', deps: [medium.id] }); assert.equal(store.getTask(dep.id)!.status, 'blocked'); const state = freshState(); const { deps } = mockDeps(state); createOrchestrator(store, noopLog, deps).claimTick(); 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('退避: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); const clock = makeClock(Date.now()); const state = freshState(); const { deps } = mockDeps(state, { nowMs: clock.nowMs }); const orch = createOrchestrator(store, noopLog, deps); orch.claimTick(); assert.equal(state.spawned.length, 0, '退避期内不领'); assert.equal(store.getTask(t.id)!.status, 'queued'); clock.advance(61_000); // 退避过期 orch.claimTick(); assert.equal(state.spawned.length, 1, '退避过期后领取'); assert.equal(store.getTask(t.id)!.status, 'executing'); store.close(); }); test('spawnWorker 抛错:兜底 failTaskAttempt(task 转 queued,executor run failed)', () => { const { store, projectId } = setup('auto-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(); 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(); }); // ───────────────────────── 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'); store.close(); }); // ───────────────────────── 多轮: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(); assert.equal(state.spawned.length, 2, '退避到期 → 重新领取 spawn'); assert.equal(store.getTask(t.id)!.status, 'executing'); store.close(); }); 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(); }); // ───────────────────────── planner:analyzing / speccing 也被领取 ───────────────────────── test('claimable 纳入 speccing:medium 任务派 planner-spec(不转状态、startRun=planner、job.runKind=planner-spec)', () => { const { store, projectId } = setup('auto-approved'); const t = store.createTask({ projectId, title: 'feat', complexity: 'medium' }); // → speccing assert.equal(store.getTask(t.id)!.status, 'speccing'); const state = freshState(); const { deps } = mockDeps(state); const orch = createOrchestrator(store, noopLog, deps); orch.claimTick(); assert.equal(state.spawned.length, 1, '派了一个 planner worker'); assert.equal(store.getTask(t.id)!.status, 'speccing', 'planner 不转任务状态(留 speccing 作规划中)'); const runs = store.listRuns(t.id); assert.equal(runs.length, 1); assert.equal(runs[0].kind, 'planner', 'run kind=planner'); assert.equal(runs[0].status, 'started'); assert.equal(state.jobs[0].runKind, 'planner-spec'); // 已在途(有 started planner run)→ 下一轮不重领 orch.claimTick(); assert.equal(state.spawned.length, 1, 'inflight 任务不被重复领取'); store.close(); }); test('claimable 纳入 analyzing:hard 任务派 planner-decompose', () => { const { store, projectId } = setup('auto-approved'); const t = store.createTask({ projectId, title: 'epic', complexity: 'hard' }); // → analyzing assert.equal(store.getTask(t.id)!.status, 'analyzing'); const state = freshState(); const { deps } = mockDeps(state); createOrchestrator(store, noopLog, deps).claimTick(); assert.equal(state.spawned.length, 1); assert.equal(store.getTask(t.id)!.status, 'analyzing', 'planner 不转状态'); assert.equal(store.listRuns(t.id)[0].kind, 'planner'); assert.equal(state.jobs[0].runKind, 'planner-decompose'); store.close(); }); test('auto-easy 不做规划:medium(speccing) 不被领取', () => { const { store, projectId } = setup('auto-easy'); const t = store.createTask({ projectId, title: 'feat', complexity: 'medium' }); // → speccing const state = freshState(); const { deps } = mockDeps(state); createOrchestrator(store, noopLog, deps).claimTick(); assert.equal(state.spawned.length, 0, 'auto-easy 只做 easy 执行,不规划 medium/hard'); assert.equal(store.getTask(t.id)!.status, 'speccing'); store.close(); }); test('并发闸用 inflight:1 个 executing + 1 个 planner 占满 concurrency=2', () => { const { store, projectId } = setup('auto-approved', 2); const e = store.createTask({ projectId, title: 'easy', complexity: 'easy' }); // → ready const m = store.createTask({ projectId, title: 'med', complexity: 'medium' }); // → speccing store.createTask({ projectId, title: 'easy2', complexity: 'easy' }); // → ready(第三个) const state = freshState(); const { deps } = mockDeps(state); createOrchestrator(store, noopLog, deps).claimTick(); // concurrency=2:一个 executor(easy) + 一个 planner(medium) 占满,第三个不领 assert.equal(state.spawned.length, 2); assert.equal(store.getTask(e.id)!.status, 'executing'); assert.equal(store.getTask(m.id)!.status, 'speccing'); store.close(); });