import { test } from 'node:test'; import assert from 'node:assert/strict'; import { mkdtempSync, rmSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { Store } from '../src/store/index.js'; import { ingestRun, shouldAutoApproveExec } from '../src/daemon/ingest.js'; import { appendOutbox, type OutboxPayload } from '../src/executor/protocol.js'; import { branchFor, worktreeDirFor } from '../src/executor/worktree.js'; const noopLog = { info: (): void => undefined, error: (): void => undefined }; /** * 隔离的 MAESTRO_DATA_DIR(outbox 落在临时目录),跑回调,结束清理。 * protocol.ts 的 runDir/outboxPath 在调用时读 env,故进 cb 前设好、cb 内的 append/ingest 都命中临时目录。 */ function withTmpDataDir(cb: () => void): void { const dir = mkdtempSync(join(tmpdir(), 'maestro-ingest-')); const prev = process.env.MAESTRO_DATA_DIR; process.env.MAESTRO_DATA_DIR = dir; try { cb(); } finally { if (prev === undefined) delete process.env.MAESTRO_DATA_DIR; else process.env.MAESTRO_DATA_DIR = prev; rmSync(dir, { recursive: true, force: true }); } } /** 真 Store + 项目 + easy 任务推进到 executing,并建一条 executor run(模拟 daemon 已领取、worker 已 spawn)。 */ function setupExecuting(opts: { autonomy?: 'auto-easy' | 'auto-approved'; maxRetries?: number } = {}): { store: Store; projectId: string; taskId: string; runId: string; dir: string; branch: string; } { const store = new Store(':memory:'); const p = store.createProject({ name: 'ingest', repoPath: '/tmp/ingest-repo-' + Math.random(), autonomy: opts.autonomy ?? 'auto-easy', maxRetries: opts.maxRetries, }); const t = store.createTask({ projectId: p.id, title: 'tweak', complexity: 'easy' }); store.setOperations(t.id, '在 README.md 追加一行'); const branch = branchFor(t.id); const dir = worktreeDirFor(p.repoPath, t.id); store.transition(t.id, 'queued', { by: 'test' }); store.transition(t.id, 'executing', { by: 'test' }); const run = store.startRun(t.id, 'executor', { worktree: dir, branch }); store.setWorkerPid(run.id, 4242); return { store, projectId: p.id, taskId: t.id, runId: run.id, dir, branch }; } function resultPayload(branch: string, worktree: string): OutboxPayload { return { type: 'result', branch, worktree, diffSummary: ' README.md | 1 +', commits: ['abc1234 hello maestro'], executor: { transcriptRef: '/tmp/exec.jsonl', sessionId: 'sess-exec-1' }, code: { summary: '## 做了什么\nmock 复审通过', verdict: 'approve', transcriptRef: '/tmp/code.jsonl' }, security: { summary: '## 安全审计\nmock 审计通过', verdict: 'approve', transcriptRef: '/tmp/sec.jsonl' }, }; } test('ingest result:task→exec_review,setResult 四字段正确,reviewer/security/executor 各 1 条 succeeded', () => { withTmpDataDir(() => { const { store, taskId, runId, dir, branch } = setupExecuting(); appendOutbox(runId, { type: 'started', pid: 4242, worktree: dir, branch, model: 'claude-sonnet-4-6' }); appendOutbox(runId, { type: 'phase', phase: 'executing' }); appendOutbox(runId, resultPayload(branch, dir)); appendOutbox(runId, { type: 'done' }); ingestRun(store, noopLog, runId); const done = store.getTask(taskId)!; assert.equal(done.status, 'exec_review'); assert.deepEqual(done.result, { branch, worktree: dir, diffSummary: ' README.md | 1 +', commits: ['abc1234 hello maestro'], prUrl: null, summary: '## 做了什么\nmock 复审通过', verdict: 'approve', securitySummary: '## 安全审计\nmock 审计通过', securityVerdict: 'approve', mergeTaskId: null, }); const runs = store.listRuns(taskId); const executor = runs.find((r) => r.kind === 'executor')!; assert.equal(executor.status, 'succeeded'); assert.equal(executor.transcriptRef, '/tmp/exec.jsonl'); assert.equal(executor.claudeSessionId, 'sess-exec-1'); const reviewer = runs.filter((r) => r.kind === 'reviewer'); assert.equal(reviewer.length, 1); assert.equal(reviewer[0].status, 'succeeded'); assert.equal(reviewer[0].transcriptRef, '/tmp/code.jsonl'); const security = runs.filter((r) => r.kind === 'security'); assert.equal(security.length, 1); assert.equal(security[0].status, 'succeeded'); assert.equal(security[0].transcriptRef, '/tmp/sec.jsonl'); // 游标推进到末条(done.seq=4) assert.equal(store.getRun(runId)!.lastSeq, 4); store.close(); }); }); test('ingest result:任一 verdict=reject → 硬闸退回重执行(不进 exec_review)', () => { withTmpDataDir(() => { const { store, taskId, runId, dir, branch } = setupExecuting(); appendOutbox(runId, { ...resultPayload(branch, dir), code: { summary: '发现问题', verdict: 'reject', transcriptRef: '/tmp/code.jsonl' }, security: { summary: '发现密钥泄露', verdict: 'reject', transcriptRef: '/tmp/sec.jsonl' }, } as OutboxPayload); ingestRun(store, noopLog, runId); const t = store.getTask(taskId)!; // verdict=reject 触发硬闸:退回重执行(queued),不进 exec_review assert.ok(t.status === 'queued' || t.status === 'failed' || t.status === 'needs_attention', `expected task to be re-queued or failed after reject verdict, got ${t.status}`); // 不应进入 exec_review assert.notEqual(t.status, 'exec_review'); store.close(); }); }); test('ingest failed:首次失败 → task 转 queued + 持久化退避(nextEligibleAt 有值)+ executor run failed', () => { withTmpDataDir(() => { const { store, taskId, runId } = setupExecuting(); appendOutbox(runId, { type: 'started', pid: 4242, worktree: '/x', branch: 'b', model: 'm' }); appendOutbox(runId, { type: 'failed', error: 'boom #1', transcriptRef: '/tmp/f.jsonl', sessionId: 'sess-f-1' }); appendOutbox(runId, { type: 'done' }); ingestRun(store, noopLog, runId); const t = store.getTask(taskId)!; assert.equal(t.status, 'queued'); // 第一次失败 → 重入队(默认 maxRetries=2) assert.ok(t.nextEligibleAt, 'nextEligibleAt 应被持久化(退避)'); assert.ok(Date.parse(t.nextEligibleAt!) > Date.now() - 1000, 'nextEligibleAt 在未来'); const executor = store.listRuns(taskId).find((r) => r.kind === 'executor')!; assert.equal(executor.status, 'failed'); assert.match(executor.error ?? '', /boom #1/); assert.equal(executor.transcriptRef, '/tmp/f.jsonl'); assert.equal(executor.claudeSessionId, 'sess-f-1'); store.close(); }); }); test('ingest failed:maxRetries=0 → 首次失败直接 needs_attention', () => { withTmpDataDir(() => { const { store, taskId, runId } = setupExecuting({ maxRetries: 0 }); appendOutbox(runId, { type: 'failed', error: 'boom', transcriptRef: null, sessionId: null }); ingestRun(store, noopLog, runId); assert.equal(store.getTask(taskId)!.status, 'needs_attention'); store.close(); }); }); test('幂等:同一批 outbox ingest 两次,DB 不重复变更(lastSeq 生效)', () => { withTmpDataDir(() => { const { store, taskId, runId, dir, branch } = setupExecuting(); appendOutbox(runId, { type: 'started', pid: 4242, worktree: dir, branch, model: 'm' }); appendOutbox(runId, resultPayload(branch, dir)); appendOutbox(runId, { type: 'done' }); ingestRun(store, noopLog, runId); const after1 = store.getTask(taskId)!; const runs1 = store.listRuns(taskId); assert.equal(after1.status, 'exec_review'); assert.equal(runs1.filter((r) => r.kind === 'reviewer').length, 1); assert.equal(runs1.filter((r) => r.kind === 'security').length, 1); const lastSeq1 = store.getRun(runId)!.lastSeq; // 第二次 ingest:seq 全部 ≤ lastSeq → 不处理任何记录 ingestRun(store, noopLog, runId); const runs2 = store.listRuns(taskId); assert.equal(runs2.length, runs1.length, '复审 run 不应重复创建'); assert.equal(runs2.filter((r) => r.kind === 'reviewer').length, 1); assert.equal(runs2.filter((r) => r.kind === 'security').length, 1); assert.equal(store.getRun(runId)!.lastSeq, lastSeq1, 'lastSeq 不变'); assert.equal(store.getTask(taskId)!.status, 'exec_review'); store.close(); }); }); test('ingest 续读:先 ingest 半截(started/phase),再追加 result,第二次 ingest 完成 exec_review', () => { withTmpDataDir(() => { const { store, taskId, runId, dir, branch } = setupExecuting(); appendOutbox(runId, { type: 'started', pid: 4242, worktree: dir, branch, model: 'm' }); appendOutbox(runId, { type: 'phase', phase: 'executing' }); ingestRun(store, noopLog, runId); assert.equal(store.getTask(taskId)!.status, 'executing'); // 还没到 result,仍 executing assert.equal(store.getRun(runId)!.lastSeq, 2); appendOutbox(runId, resultPayload(branch, dir)); appendOutbox(runId, { type: 'done' }); ingestRun(store, noopLog, runId); assert.equal(store.getTask(taskId)!.status, 'exec_review'); assert.equal(store.getRun(runId)!.lastSeq, 4); store.close(); }); }); // ───────────────────────── planner run 摄取 ───────────────────────── /** 真 Store + 一个 medium(speccing) 或 hard(analyzing) 任务 + 一条 planner run(模拟已领取规划)。 */ function setupPlanning(complexity: 'medium' | 'hard'): { store: Store; taskId: string; runId: string } { const store = new Store(':memory:'); const p = store.createProject({ name: 'plan', repoPath: '/tmp/plan-' + Math.random(), autonomy: 'auto-approved' }); const t = store.createTask({ projectId: p.id, title: '大任务', complexity }); // 建任务即落 speccing(medium)/analyzing(hard),不转状态,直接起 planner run const run = store.startRun(t.id, 'planner', { worktree: p.repoPath, branch: 'n/a' }); store.setWorkerPid(run.id, 4243); return { store, taskId: t.id, runId: run.id }; } test('ingest planner-spec:setSpec + 转 spec_review + 收尾 planner run', () => { withTmpDataDir(() => { const { store, taskId, runId } = setupPlanning('medium'); assert.equal(store.getTask(taskId)!.status, 'speccing'); appendOutbox(runId, { type: 'started', pid: 4243, worktree: '/r', branch: 'n/a', model: 'opus' }); appendOutbox(runId, { type: 'spec-result', spec: '## 改动\n改 a.ts', transcriptRef: '/p.jsonl', sessionId: 's' }); appendOutbox(runId, { type: 'done' }); ingestRun(store, noopLog, runId); const t = store.getTask(taskId)!; assert.equal(t.status, 'spec_review'); assert.equal(t.spec, '## 改动\n改 a.ts'); assert.equal(store.getRun(runId)!.status, 'succeeded'); store.close(); }); }); test('ingest planner-decompose:setPlan + 建子任务 + 转 plan_review', () => { withTmpDataDir(() => { const { store, taskId, runId } = setupPlanning('hard'); assert.equal(store.getTask(taskId)!.status, 'analyzing'); appendOutbox(runId, { type: 'decompose-result', plan: '拆成两步', subtasks: [{ title: '建骨架', complexity: 'easy' }, { title: '实现', complexity: 'medium' }], transcriptRef: null, sessionId: null, }); appendOutbox(runId, { type: 'done' }); ingestRun(store, noopLog, runId); const t = store.getTask(taskId)!; assert.equal(t.status, 'plan_review'); assert.equal(t.plan, '拆成两步'); const kids = store.listTasks(t.projectId).filter((x) => x.parentId === taskId); assert.equal(kids.length, 2, '建了 2 个子任务'); assert.deepEqual(kids.map((k) => k.complexity).sort(), ['easy', 'medium']); assert.equal(store.getRun(runId)!.status, 'succeeded'); store.close(); }); }); test('autoApprovePlan:auto-approved + 开关 + 全 easy + 改动面小 → 跳过 plan_review 直接 decomposed', () => { withTmpDataDir(() => { const { store, taskId, runId } = setupPlanning('hard'); const pid = store.getTask(taskId)!.projectId; store.patchProject(pid, { autoApprovePlan: true }); appendOutbox(runId, { type: 'decompose-result', plan: '拆成两步', subtasks: [{ title: 'a', complexity: 'easy' }, { title: 'b', complexity: 'easy' }], transcriptRef: null, sessionId: null, }); appendOutbox(runId, { type: 'done' }); ingestRun(store, noopLog, runId); assert.equal(store.getTask(taskId)!.status, 'decomposed', '全 easy 自动放行'); store.close(); }); }); test('autoApprovePlan:开关关 → 仍走 plan_review 人审', () => { withTmpDataDir(() => { const { store, taskId, runId } = setupPlanning('hard'); // autoApprovePlan 默认 false appendOutbox(runId, { type: 'decompose-result', plan: 'x', subtasks: [{ title: 'a', complexity: 'easy' }], transcriptRef: null, sessionId: null, }); appendOutbox(runId, { type: 'done' }); ingestRun(store, noopLog, runId); assert.equal(store.getTask(taskId)!.status, 'plan_review', '开关关 → 人审'); store.close(); }); }); test('autoApprovePlan:含 medium 子任务 → 不放行(改动面/复杂度不满足)', () => { withTmpDataDir(() => { const { store, taskId, runId } = setupPlanning('hard'); const pid = store.getTask(taskId)!.projectId; store.patchProject(pid, { autoApprovePlan: true }); appendOutbox(runId, { type: 'decompose-result', plan: 'x', subtasks: [{ title: 'a', complexity: 'easy' }, { title: 'b', complexity: 'medium' }], transcriptRef: null, sessionId: null, }); appendOutbox(runId, { type: 'done' }); ingestRun(store, noopLog, runId); assert.equal(store.getTask(taskId)!.status, 'plan_review', '有 medium → 人审'); store.close(); }); }); test('shouldAutoApproveExec:双 approve + auto-approved + 开关 → true;任一不满足 → false', () => { const s = new Store(':memory:'); const p = s.createProject({ name: 'aae', repoPath: '/tmp/aae-' + Math.random(), autonomy: 'auto-approved' }); const t = s.createTask({ projectId: p.id, title: 'x', complexity: 'easy' }); // 开关默认关 → false assert.equal(shouldAutoApproveExec(s, t, 'approve', 'approve'), false, '开关关'); s.patchProject(p.id, { autoApproveExec: true }); // 现在满足三条件 → true(函数读 task.projectId 自取项目,传错 id 不可能) assert.equal(shouldAutoApproveExec(s, t, 'approve', 'approve'), true); // verdict 任一非 approve → false assert.equal(shouldAutoApproveExec(s, t, 'approve', 'reject'), false); assert.equal(shouldAutoApproveExec(s, t, null, 'approve'), false); // 非 auto-approved → false s.patchProject(p.id, { autonomy: 'auto-easy' }); assert.equal(shouldAutoApproveExec(s, t, 'approve', 'approve'), false); s.close(); }); test('ingest planner failed:走 failPlanAttempt(退避,任务留 speccing,不进 failed)', () => { withTmpDataDir(() => { const { store, taskId, runId } = setupPlanning('medium'); appendOutbox(runId, { type: 'failed', error: 'planner 崩了', transcriptRef: null, sessionId: null }); appendOutbox(runId, { type: 'done' }); ingestRun(store, noopLog, runId); const t = store.getTask(taskId)!; assert.equal(t.status, 'speccing', '默认 maxRetries=2,第一次失败留 speccing 退避重试'); assert.ok(t.nextEligibleAt, '设了退避 nextEligibleAt'); assert.equal(store.getRun(runId)!.status, 'failed'); store.close(); }); });