phase3(集成测试): 真 worker 进程 spawn → 重启 re-adopt 存活 worker → ingest;死亡→回收重试

用假 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 <noreply@anthropic.com>

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/<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);
+});

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