Files
maestro/test/ingest.test.ts
T
wangjia 73f80d27c9 feat(agent): 接通 plan 自动放行(autoApprovePlan)+ 项目配置可写
之前 auto_approve_plan/auto_approve_exec/checks 列与 Project 类型已存在,但
PatchProjectInput 不含它们 → 无法启用(dead config)。本次接通:
- PatchProjectInput + patchProject 支持 autoApprovePlan/autoApproveExec/checks
- API PATCH /projects 透传三字段
- ingest decompose-result:auto-approved + autoApprovePlan + 全 easy + 子任务≤5
  → store.decide(accept) 跳过 plan_review 直达 decomposed(默认关)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-24 22:13:18 +08:00

320 lines
14 KiB
TypeScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
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 } 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_DIRoutbox 落在临时目录),跑回调,结束清理。
* 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 resulttask→exec_reviewsetResult 四字段正确,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 failedmaxRetries=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;
// 第二次 ingestseq 全部 ≤ 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-specsetSpec + 转 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-decomposesetPlan + 建子任务 + 转 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('autoApprovePlanauto-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('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();
});
});