wangjia
027a9cb2d6
- pipeline.test.ts：PipelineDeps 增加 syncMain 可注入接口（默认用真 git）， 测试 mockDeps 加 syncMain=no-op + runConflict，消除 fake-worktree 上真 git 调用失败 - models.test.ts：reviewer 三档默认 fable-5、project.model 不覆盖复审/conflict、 MODEL_FALLBACK_CHAIN 三项链；pickFallbackModel 改为取链上下一档（循环）而非首个不等项 - ingest.test.ts：verdict=reject 硬闸测试改为断言未进 exec_review Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
271 lines
12 KiB
TypeScript
271 lines
12 KiB
TypeScript
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_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('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();
|
||
});
|
||
});
|