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 { 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'); });