From 70d996110da9a7f9139d6190657713e641a2cbfd Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Sat, 13 Jun 2026 13:07:42 +0800 Subject: [PATCH] =?UTF-8?q?phase3(=E9=9B=86=E6=88=90=E6=B5=8B=E8=AF=95):?= =?UTF-8?q?=20=E7=9C=9F=20worker=20=E8=BF=9B=E7=A8=8B=20spawn=20=E2=86=92?= =?UTF-8?q?=20=E9=87=8D=E5=90=AF=20re-adopt=20=E5=AD=98=E6=B4=BB=20worker?= =?UTF-8?q?=20=E2=86=92=20ingest;=E6=AD=BB=E4=BA=A1=E2=86=92=E5=9B=9E?= =?UTF-8?q?=E6=94=B6=E9=87=8D=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 用假 worker(走 outbox 协议、不烧 SDK token)验证 daemon↔worker 真进程 seam: - 头条: spawn 真进程→等 started→换 Store 重连 reconcile(真判活)→re-adopt(任务仍 executing 不重跑) →finish 哨兵→ingest 到 exec_review(四字段 + reviewer/security/executor run 就位) - 回收: 杀进程组 + 心跳调陈旧→reconcile 回收→failTaskAttempt 退避重入队 Co-Authored-By: Claude Opus 4.8 --- test/fixtures/fake-worker.ts | 37 +++++++++ test/worker-integration.test.ts | 140 ++++++++++++++++++++++++++++++++ 2 files changed, 177 insertions(+) create mode 100644 test/fixtures/fake-worker.ts create mode 100644 test/worker-integration.test.ts diff --git a/test/fixtures/fake-worker.ts b/test/fixtures/fake-worker.ts new file mode 100644 index 0000000..bd9ea00 --- /dev/null +++ b/test/fixtures/fake-worker.ts @@ -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//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); +}); diff --git a/test/worker-integration.test.ts b/test/worker-integration.test.ts new file mode 100644 index 0000000..5470910 --- /dev/null +++ b/test/worker-integration.test.ts @@ -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 { + 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'); +});