maestro/test/orchestrator.test.ts

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<OrchestratorDeps> = {}): { 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();
});