maestro/test/worker-integration.test.ts

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