wangjia/maestro
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: PR 审核流程 + 自动合并 + macOS 通知 + 订阅额度透传 + 归档详情

Browse Source 
PR 流程（开发→提PR→CodeReview→安全审计→审核→通过即合并）：
- reviewer 拆分为 code review 与安全审计两个独立 CC run（kind=reviewer/security），
  TaskResult 四字段(summary/verdict + securitySummary/securityVerdict)，闸上两节报告两枚结论徽章
- executor/merge.ts: 通过即合并——临时 worktree 内 merge --no-ff，绝不碰用户工作区/不 push；
  冲突安全拒绝(报冲突文件)；重复合并幂等；合并后回收执行 worktree + 删分支
- decide 路由 merge:false 逃生口 + 看板「仅通过」按钮(目标分支被工作区检出时用)

通知（daemon/notify.ts）：
- macOS 原生通知: 进审核闸(复审建议拒绝标⚠)/连续失败需人工/合并完成
- 同任务同类型 60s 抑制、osascript 转义截断、MAESTRO_NOTIFY=0 关闭

订阅额度透传（daemon/usage.ts）：
- OAuth usage API(与 Claude Code/claude-hud 同源)，凭证 keychain→内存零泄漏
- 60s 成败双缓存+并发去重+5s 超时，失败降级 null
- GET /api/agents 顶层 usage 字段；Agent 面板显示 5h/周用量条+重置倒计时(>80%琥珀/>95%红)

看板与生命周期：
- 归档区: 深度1整树完成沉底，时间倒序分页(尺寸 chip 10/20/50/100 置底)
- 归档详情对话框: 全属性/执行历史与时长/审批记录/状态流转时间线(GET /api/tasks/:id/events)
- 容器收口: 已拆解 Hard 子任务全 done 自动 done(afterDone 逐级向上)
- 同步按钮收进配置面板；执行白名单扩测试命令(npm/go/shellcheck/make/pytest)
- Agent 面板显示调度模式与各复杂度模型；被依赖阻塞→被阻塞

测试: 74/74(新增 merge 6/notify 10/usage 7/容器收口/双复审适配)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
wangjia
2026-06-13 03:22:30 +08:00
parent f18db021c3
commit 740d2c2637
18 changed files with 1102 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/server.ts
+42 -4
View File
@@ -7,6 +7,9 @@ import type { TaskStatus } from '../model/status.js';
import type { Project, Autonomy } from '../model/types.js';
import { syncProject, hasTodoJson } from '../sync/todo-sync.js';
import { resolvedExecutorModels } from '../executor/models.js';
import { mergeBranch } from '../executor/merge.js';
import { git, removeWorktree } from '../executor/worktree.js';
import { createUsageFetcher, type UsageInfo } from '../daemon/usage.js';
/** Project 出参：附加 hasTodoJson（<repoPath>/todo/todo.json 是否存在，每次序列化时算） */
function projectOut(p: Project): Project & { hasTodoJson: boolean } {
@@ -16,6 +19,8 @@ function projectOut(p: Project): Project & { hasTodoJson: boolean } {
export interface ApiOptions {
store: Store;
logger?: boolean;
/** Claude 订阅额度查询（测试注入；默认直连 OAuth usage API，60s 缓存） */
getUsage?: () => Promise<UsageInfo | null>;
}
/**
@@ -75,7 +80,9 @@ export function buildServer(opts: ApiOptions): FastifyInstance {
});
// ---------- Agents（每项目一条；active 来自 runs 表 status='started'，执行器 Phase 2 前通常为空） ----------
app.get('/api/agents', () => {
// usage = Claude 订阅额度（执行 agent 烧的就是这个池子）；查询失败 → null，前端降级显示
const getUsage = opts.getUsage ?? createUsageFetcher({ log: app.log });
app.get('/api/agents', async () => {
const runs = store.activeRuns();
const byProject = new Map<string, ActiveRun[]>();
for (const r of runs) {
@@ -95,7 +102,7 @@ export function buildServer(opts: ApiOptions): FastifyInstance {
runId: r.runId, taskId: r.taskId, taskTitle: r.taskTitle, kind: r.kind, startedAt: r.startedAt,
})),
}));
return { totalActive: runs.length, agents };
return { totalActive: runs.length, agents, usage: await getUsage() };
});
app.get('/api/projects/:id/tasks', (req) => {
@@ -197,10 +204,41 @@ export function buildServer(opts: ApiOptions): FastifyInstance {
});
// 审批闸：accept / reject（reject 必带 reason）
app.post('/api/tasks/:id/decide', (req) => {
// exec 闸的 accept = 通过并合并（PR 闭环）：先 merge 再 decide；merge 失败 → 400，任务保留在审核闸。
app.post('/api/tasks/:id/decide', async (req) => {
const { id } = req.params as { id: string };
const b = req.body as { action?: string; actor?: string; reason?: string | null };
const b = req.body as { action?: string; actor?: string; reason?: string | null; merge?: boolean };
if (b?.action !== 'accept' && b?.action !== 'reject') throw new StoreError('action 必须是 accept|reject');
const task = store.getTask(id);
// merge:false = 仅通过不合并（逃生口：如目标分支正被用户工作区检出导致自动合并不可用）
if (task && task.status === 'exec_review' && b.action === 'accept' && task.result?.branch && b.merge !== false) {
const project = store.getProject(task.projectId);
if (!project) throw new StoreError(`项目不存在: ${task.projectId}`);
const { branch, worktree } = task.result;
const mr = await mergeBranch(project.repoPath, branch, project.defaultBranch, task.id);
if (!mr.ok) throw new StoreError(`合并失败：${mr.error}（任务保留在审核闸）`);
store.decide(id, 'accept', b.actor ?? 'user', b.reason ?? null);
// 合并产物记录：复用 prUrl 字段写 merged:<mergeCommit>
store.setResult(id, { ...task.result, prUrl: `merged:${mr.mergeCommit}` });
// 异步回收：执行 worktree + 已合并的任务分支（失败只记日志，不影响响应）
void (async () => {
try {
if (worktree) await removeWorktree(project.repoPath, worktree);
} catch (e) {
app.log.error(`任务 ${id} 合并后清理 worktree 失败（不影响结果）：${(e as Error).message}`);
}
try {
await git(project.repoPath, ['branch', '-d', branch]);
} catch (e) {
app.log.error(`任务 ${id} 合并后删除分支 ${branch} 失败（不影响结果）：${(e as Error).message}`);
}
})();
return store.getTask(id);
}
// 非 exec 闸（或 reject / 无分支结果）：行为完全不变
return store.decide(id, b.action, b.actor ?? 'user', b.reason ?? null);
});
src/daemon/index.ts
+3
View File
@@ -5,6 +5,7 @@ import { buildServer, attachWebSocket } from '../api/server.js';
import { registerStatic } from '../api/static.js';
import { loadConfig } from './config.js';
import { startOrchestrator } from './orchestrator.js';
import { startNotifier } from './notify.js';
import { syncProject, todoJsonPath } from '../sync/todo-sync.js';
/**
@@ -65,11 +66,13 @@ async function main(): Promise<void> {
const syncTimer = startSyncLoop(store, app);
const orchTimer = startOrchestrator(store, app); // 编排器：自动领取可执行任务（MAESTRO_ORCH_INTERVAL 秒，0=关闭）
const stopNotifier = startNotifier(store, app); // macOS 原生通知：审核闸/需人工/合并完成（MAESTRO_NOTIFY=0 关闭）
const shutdown = async (): Promise<void> => {
app.log.info('收到退出信号，关闭中…');
if (syncTimer) clearInterval(syncTimer);
if (orchTimer) clearInterval(orchTimer);
if (stopNotifier) stopNotifier();
await app.close();
store.close();
process.exit(0);
src/daemon/notify.ts
+114
View File
@@ -0,0 +1,114 @@
import { execFile } from 'node:child_process';
import type { Store } from '../store/index.js';
import type { Event } from '../model/types.js';
export interface NotifyLogger {
info(msg: string): void;
error(msg: string): void;
}
/** osascript 执行函数（测试注入假实现，不真发通知） */
export type OsascriptFn = (script: string) => void;
/** 同一任务同类型通知的抑制窗口 */
export const NOTIFY_SUPPRESS_MS = 60_000;
const MAX_LEN = 80;
/** AppleScript 字符串注入防护：去换行、转义反斜杠与双引号，截断 80 字符 */
export function sanitizeForOsascript(s: string): string {
const oneLine = String(s).replace(/[\r\n]+/g, ' ').trim();
const cut = oneLine.length > MAX_LEN ? `${oneLine.slice(0, MAX_LEN - 1)}…` : oneLine;
return cut.replace(/\\/g, '\\\\').replace(/"/g, '\\"');
}
const GATE_NAME: Record<string, string> = { plan_review: '拆解', spec_review: '方案', exec_review: 'PR' };
interface NotifierOptions {
osascript?: OsascriptFn;
now?: () => number;
}
/**
* 事件 → 通知文案（不触发的事件返回 null）。kind 用于同任务同类型 60s 抑制。
* - status.changed → plan/spec/exec_review：「⏳ <任务名> 等待审核（拆解/方案/PR）」；
* exec_review 且复审任一 verdict=reject：「⚠ <任务名> 复审建议拒绝（PR 待审核）」
* - status.changed → needs_attention：「❌ <任务名> 连续失败需人工」
* - approval.granted 且 gate=exec：「✅ <任务名> 已合并完成」
*/
export function buildNotification(store: Store, e: Event): { kind: string; message: string } | null {
if (!e.taskId) return null;
if (e.type === 'status.changed') {
const to = String(e.payload.to ?? '');
if (to === 'needs_attention') {
const t = store.getTask(e.taskId);
if (!t) return null;
return { kind: 'attention', message: `❌ ${t.title} 连续失败需人工` };
}
if (to === 'plan_review' || to === 'spec_review' || to === 'exec_review') {
const t = store.getTask(e.taskId);
if (!t) return null;
if (to === 'exec_review' && (t.result?.verdict === 'reject' || t.result?.securityVerdict === 'reject')) {
return { kind: 'gate', message: `⚠ ${t.title} 复审建议拒绝（PR 待审核）` };
}
return { kind: 'gate', message: `⏳ ${t.title} 等待审核（${GATE_NAME[to]}）` };
}
return null;
}
if (e.type === 'approval.granted' && e.payload.gate === 'exec') {
const t = store.getTask(e.taskId);
if (!t) return null;
return { kind: 'merged', message: `✅ ${t.title} 已合并完成` };
}
return null;
}
/**
* 创建通知器：订阅 store 事件流，符合条件时发 macOS 原生通知（osascript display notification）。
* 同一任务同类型 60s 内只发一次；osascript 失败只记日志。返回取消订阅函数。
*/
export function createNotifier(store: Store, log: NotifyLogger, opts: NotifierOptions = {}): () => void {
const now = opts.now ?? Date.now;
const osa: OsascriptFn = opts.osascript ?? ((script) => {
execFile('osascript', ['-e', script], (err) => {
if (err) log.error(`macOS 通知发送失败：${err.message}`);
});
});
const lastSent = new Map<string, number>(); // `${taskId}:${kind}` → 最近发送时刻
return store.subscribe((e) => {
try {
const n = buildNotification(store, e);
if (!n) return;
const key = `${e.taskId}:${n.kind}`;
const t = now();
const prev = lastSent.get(key);
if (prev !== undefined && t - prev < NOTIFY_SUPPRESS_MS) return;
lastSent.set(key, t);
osa(`display notification "${sanitizeForOsascript(n.message)}" with title "maestro"`);
} catch (err) {
log.error(`通知处理失败：${(err as Error).message}`);
}
});
}
/**
* 接线入口（照 startSyncLoop 模式）：MAESTRO_NOTIFY=0 关闭；非 macOS 平台不启用。
* 返回取消订阅函数（shutdown 时调用），未启用时返回 null。
*/
export function startNotifier(store: Store, app: { log: NotifyLogger }): (() => void) | null {
if (process.env.MAESTRO_NOTIFY === '0') {
app.log.info('macOS 通知已关闭（MAESTRO_NOTIFY=0）');
return null;
}
if (process.platform !== 'darwin') {
app.log.info('非 macOS 平台，原生通知不启用');
return null;
}
const stop = createNotifier(store, app.log);
app.log.info('macOS 通知已启用：审核闸/需人工/合并完成（同任务同类型 60s 抑制，MAESTRO_NOTIFY=0 关闭）');
return stop;
}
src/daemon/orchestrator.ts
+42 -30
View File
@@ -4,7 +4,7 @@ import { rankByScore } from '../model/scoring.js';
import { createWorktree, worktreeDiff, type WorktreeDiff, type WorktreeInfo } from '../executor/worktree.js';
import { runTask, type RunnerFn } from '../executor/runner.js';
import { runVerify, type VerifyFn } from '../executor/verify.js';
import { reviewTask, type ReviewerFn } from '../executor/reviewer.js';
import { reviewCode, reviewSecurity, type ReviewerFn } from '../executor/reviewer.js';
import { pickModel } from '../executor/models.js';
/** 失败后最多自动重试次数（重试 2 次 = 最多 3 次执行），之后 → needs_attention */
@@ -18,7 +18,8 @@ export interface OrchestratorLogger {
/** 依赖注入点：测试传 mock，生产用真实现 */
export interface OrchestratorDeps {
runner: RunnerFn;
reviewer: ReviewerFn;
reviewCode: ReviewerFn; // code review（kind=reviewer 的 run）
reviewSecurity: ReviewerFn; // 安全审计（kind=security 的 run）
verify: VerifyFn;
createWorktree: (repoPath: string, taskId: string, baseBranch: string) => Promise<WorktreeInfo>;
worktreeDiff: (repoPath: string, dir: string, branch: string, baseBranch: string) => Promise<WorktreeDiff>;
@@ -40,7 +41,7 @@ export interface Orchestrator {
* 成功 → setResult + exec_review；失败 → failed → 重试 ≤MAX_RETRIES 次 → needs_attention。
*/
export function createOrchestrator(store: Store, log: OrchestratorLogger, deps: Partial<OrchestratorDeps> = {}): Orchestrator {
const d: OrchestratorDeps = { runner: runTask, reviewer: reviewTask, verify: runVerify, createWorktree, worktreeDiff, ...deps };
const d: OrchestratorDeps = { runner: runTask, reviewCode, reviewSecurity, verify: runVerify, createWorktree, worktreeDiff, ...deps };
const inflight = new Map<string, string>(); // taskId → projectId
const pending = new Set<Promise<void>>();
@@ -63,7 +64,39 @@ export function createOrchestrator(store: Store, log: OrchestratorLogger, deps:
return rankByScore(candidates, tasks);
}
/** 单任务全流程：executing → worktree → run → verify → reviewer(复审，失败不挡) → setResult(summary/verdict) → exec_review / failed(重试) */
/** 跑一个独立复审 run（kind=reviewer/security）。任何失败折叠为 summary=失败原因、verdict=null，不抛错、不挡任务。 */
async function runOneReview(
kind: 'reviewer' | 'security',
fn: ReviewerFn,
task: Task,
project: Project,
wt: WorktreeInfo,
executorReport: string,
): Promise<{ summary: string | null; verdict: ReviewVerdict | null }> {
const label = kind === 'reviewer' ? 'code review' : '安全审计';
let reviewRunId: string | null = null;
try {
const review = store.startRun(task.id, kind, { worktree: wt.dir, branch: wt.branch });
reviewRunId = review.id;
log.info(`${label} 任务 ${task.id} run=${review.id} model=${pickModel(task, project, 'reviewer')}`);
const rv = await fn(task, project, wt, review.id, executorReport);
store.finishRun(review.id, 'succeeded', {
transcriptRef: rv.transcriptRef ?? undefined,
claudeSessionId: rv.sessionId ?? undefined,
});
log.info(`任务 ${task.id} ${label} 完成 verdict=${rv.verdict ?? '（未解析到）'}`);
return { summary: rv.summary, verdict: rv.verdict };
} catch (e) {
const reMsg = (e as Error).message;
if (reviewRunId) {
try { store.finishRun(reviewRunId, 'failed', { error: reMsg }); } catch { /* 收尾失败不影响主流程 */ }
}
log.error(`任务 ${task.id} ${label} 失败（不挡任务，照常进 exec_review）：${reMsg}`);
return { summary: `自动复审失败：${reMsg}`, verdict: null };
}
}
/** 单任务全流程：executing → worktree → run → verify → code review run → 安全审计 run（均失败不挡）→ setResult(四字段) → exec_review / failed(重试) */
async function executeTask(project: Project, task: Task): Promise<void> {
let runId: string | null = null;
let runClosed = false;
@@ -98,36 +131,15 @@ export function createOrchestrator(store: Store, log: OrchestratorLogger, deps:
const diff = await d.worktreeDiff(project.repoPath, wt.dir, wt.branch, project.defaultBranch);
// 自动复审（kind=reviewer 的新 run）：失败不挡任务，summary 记失败原因、verdict=null
let summary: string | null = null;
let verdict: ReviewVerdict | null = null;
let reviewRunId: string | null = null;
try {
const review = store.startRun(task.id, 'reviewer', { worktree: wt.dir, branch: wt.branch });
reviewRunId = review.id;
log.info(`复审任务 ${task.id} run=${review.id} model=${pickModel(task, project, 'reviewer')}`);
const rv = await d.reviewer(task, project, wt, review.id, rr.finalText ?? '');
summary = rv.summary;
verdict = rv.verdict;
store.finishRun(review.id, 'succeeded', {
transcriptRef: rv.transcriptRef ?? undefined,
claudeSessionId: rv.sessionId ?? undefined,
});
log.info(`任务 ${task.id} 复审完成 verdict=${verdict ?? '（未解析到）'}`);
} catch (e) {
const reMsg = (e as Error).message;
summary = `自动复审失败：${reMsg}`;
verdict = null;
if (reviewRunId) {
try { store.finishRun(reviewRunId, 'failed', { error: reMsg }); } catch { /* 收尾失败不影响主流程 */ }
}
log.error(`任务 ${task.id} 复审失败（不挡任务，照常进 exec_review）：${reMsg}`);
}
// 双复审（顺序）：code review run（kind=reviewer）→ 安全审计 run（kind=security）。任一失败不挡任务。
const code = await runOneReview('reviewer', d.reviewCode, task, project, wt, rr.finalText ?? '');
const sec = await runOneReview('security', d.reviewSecurity, task, project, wt, rr.finalText ?? '');
store.setResult(task.id, {
branch: wt.branch, worktree: wt.dir,
diffSummary: diff.diffSummary, commits: diff.commits, prUrl: null,
summary, verdict,
summary: code.summary, verdict: code.verdict,
securitySummary: sec.summary, securityVerdict: sec.verdict,
});
store.transition(task.id, 'exec_review', { by: 'orchestrator', runId: run.id });
store.finishRun(run.id, 'succeeded', {
src/daemon/usage.ts
+152
View File
@@ -0,0 +1,152 @@
import { readFile } from 'node:fs/promises';
import { execFile } from 'node:child_process';
import { homedir } from 'node:os';
import { join } from 'node:path';
/**
* Claude 订阅额度（5 小时窗口 + 周窗口）透传。
*
* 数据源：Anthropic OAuth usage API（GET https://api.anthropic.com/api/oauth/usage，
* Bearer = Claude Code 的 OAuth accessToken）。这正是 Claude Code 喂给 statusline
* `rate_limits` 字段的同一份数据（claude-hud 的 Usage/Weekly 即来源于此）；daemon
* 不在 Claude Code 会话内拿不到 stdin，故直连同一端点。
*
* 凭证：优先 ~/.claude/.credentials.json（Linux），其次 macOS keychain
* （service "Claude Code-credentials"）。token 只进内存，绝不写日志/响应。
*/
export interface UsageWindow {
percent: number; // 0-100 已用百分比
resetsAt: string | null; // ISO 时间，窗口重置时刻
}
export interface UsageInfo {
session: UsageWindow | null; // 5 小时滚动窗口
weekly: UsageWindow | null; // 7 天窗口
}
export interface UsageLogger {
info(msg: string): void;
error(msg: string): void;
}
export const USAGE_CACHE_MS = 60_000;
export const USAGE_ENDPOINT = 'https://api.anthropic.com/api/oauth/usage';
const FETCH_TIMEOUT_MS = 5_000;
export interface UsageDeps {
/** 取 OAuth accessToken（测试注入；返回 null = 凭证不可用）。绝不把返回值写进日志。 */
readToken?: () => Promise<string | null>;
fetchFn?: typeof fetch;
now?: () => number;
log?: UsageLogger;
}
/** 解析凭证 JSON（~/.claude/.credentials.json 与 keychain 存的是同一结构） */
function tokenFromCredentialJson(raw: string, now: number, log: UsageLogger): string | null {
const j = JSON.parse(raw) as { claudeAiOauth?: { accessToken?: string; expiresAt?: number } };
const oauth = j.claudeAiOauth;
if (!oauth?.accessToken) return null;
if (typeof oauth.expiresAt === 'number' && oauth.expiresAt <= now) {
log.info('额度查询：OAuth token 已过期（等 Claude Code 刷新后自动恢复）');
return null;
}
return oauth.accessToken;
}
/** 默认凭证读取：credentials 文件 → macOS keychain。任何失败返回 null。 */
async function defaultReadToken(now: number, log: UsageLogger): Promise<string | null> {
const configDir = process.env.CLAUDE_CONFIG_DIR || join(homedir(), '.claude');
try {
const raw = await readFile(join(configDir, '.credentials.json'), 'utf8');
const tok = tokenFromCredentialJson(raw, now, log);
if (tok) return tok;
} catch {
// 文件不存在/不可解析 → 尝试 keychain
}
if (process.platform !== 'darwin') return null;
try {
const raw = await new Promise<string>((resolve, reject) => {
execFile(
'security',
['find-generic-password', '-s', 'Claude Code-credentials', '-w'],
{ timeout: 3_000 },
(err, stdout) => (err ? reject(err) : resolve(stdout)),
);
});
return tokenFromCredentialJson(raw.trim(), now, log);
} catch {
return null;
}
}
/** API 响应里的一个窗口 → UsageWindow（缺失/畸形 → null） */
function parseWindow(v: unknown): UsageWindow | null {
if (!v || typeof v !== 'object') return null;
const w = v as { utilization?: unknown; resets_at?: unknown };
if (typeof w.utilization !== 'number' || !Number.isFinite(w.utilization)) return null;
return {
percent: Math.round(Math.min(100, Math.max(0, w.utilization))),
resetsAt: typeof w.resets_at === 'string' ? w.resets_at : null,
};
}
const noopLog: UsageLogger = { info: () => undefined, error: () => undefined };
/**
* 创建 getUsage：60s 内存缓存（成功与失败都缓存，杜绝重试风暴），并发去重（in-flight 共享）。
* 任何失败（无凭证 / 网络 / 非 200 / 解析失败）→ null + 记日志，绝不抛出。
*/
export function createUsageFetcher(deps: UsageDeps = {}): () => Promise<UsageInfo | null> {
const now = deps.now ?? Date.now;
const fetchFn = deps.fetchFn ?? fetch;
const log = deps.log ?? noopLog;
const readToken = deps.readToken ?? ((): Promise<string | null> => defaultReadToken(now(), log));
let cache: { at: number; value: UsageInfo | null } | null = null;
let inflight: Promise<UsageInfo | null> | null = null;
async function fetchOnce(): Promise<UsageInfo | null> {
const token = await readToken();
if (!token) {
log.info('额度查询：未取到 Claude Code OAuth 凭证，跳过');
return null;
}
const res = await fetchFn(USAGE_ENDPOINT, {
headers: {
Authorization: `Bearer ${token}`,
'anthropic-beta': 'oauth-2025-04-20',
'Content-Type': 'application/json',
},
signal: AbortSignal.timeout(FETCH_TIMEOUT_MS),
});
if (!res.ok) {
log.error(`额度查询：usage API 返回 ${res.status}`);
return null;
}
const body = (await res.json()) as { five_hour?: unknown; seven_day?: unknown };
const session = parseWindow(body.five_hour);
const weekly = parseWindow(body.seven_day);
if (!session && !weekly) {
log.error('额度查询：usage API 响应缺少 five_hour/seven_day 字段');
return null;
}
return { session, weekly };
}
return async function getUsage(): Promise<UsageInfo | null> {
if (cache && now() - cache.at < USAGE_CACHE_MS) return cache.value;
if (inflight) return inflight;
inflight = fetchOnce()
.catch((e: unknown) => {
log.error(`额度查询失败：${(e as Error).message}`); // 错误信息不含 token
return null;
})
.then((value) => {
cache = { at: now(), value };
inflight = null;
return value;
});
return inflight;
};
}
src/executor/merge.ts
+84
View File
@@ -0,0 +1,84 @@
import { mkdirSync, rmSync } from 'node:fs';
import { dirname, join } from 'node:path';
import { git, worktreeBase } from './worktree.js';
export interface MergeResult {
ok: boolean;
mergeCommit?: string; // 成功时为合并后 defaultBranch 的 HEAD（重复合并时为当前 HEAD，不新建提交）
error?: string;
}
/** 临时合并 worktree 目录（确定性）：<dataDir>/worktrees/_merge/<taskId>/ */
export function mergeWorktreeDirFor(taskId: string): string {
return join(worktreeBase(), '_merge', taskId);
}
/**
* 把任务分支合并进 defaultBranch（PR 闭环的"merge"步）。
* 绝不触碰用户工作区：在 <dataDir>/worktrees/_merge/<taskId>/ 临时 git worktree 检出 defaultBranch，
* 在其中 `git merge --no-ff <branch>`，成功后删除临时 worktree。不 push。
* 全部 execFile('git')（见 worktree.ts 的 git()），不走 shell。
*
* 边界：
* - 分支不存在（已删/已清理）→ {ok:false, error}
* - defaultBranch 正被用户工作区检出 → git worktree add 失败 → {ok:false, error}（宁可失败也不动用户检出）
* - 冲突 → `git merge --abort` 后清理临时 worktree，error 含冲突文件列表；defaultBranch 与任务分支均无损
* - 重复合并（分支已在 defaultBranch 里）→ git 返回 Already up to date，ok:true、不新建提交
*/
export async function mergeBranch(
repoPath: string,
branch: string,
defaultBranch: string,
taskId: string,
): Promise<MergeResult> {
// 0) 分支必须存在
try {
await git(repoPath, ['rev-parse', '--verify', '--quiet', `refs/heads/${branch}`]);
} catch {
return { ok: false, error: `任务分支不存在：${branch}（可能已被删除或清理）` };
}
// 1) 清理残留的临时合并 worktree（上次中断/失败遗留），再新建
const dir = mergeWorktreeDirFor(taskId);
await git(repoPath, ['worktree', 'prune']).catch(() => undefined);
await git(repoPath, ['worktree', 'remove', '--force', dir]).catch(() => undefined);
rmSync(dir, { recursive: true, force: true });
mkdirSync(dirname(dir), { recursive: true });
try {
await git(repoPath, ['worktree', 'add', dir, defaultBranch]);
} catch (e) {
const msg = (e as Error).message;
if (/already checked out|already used by worktree/i.test(msg)) {
return {
ok: false,
error: `默认分支 ${defaultBranch} 正被某个工作区检出，无法在临时 worktree 合并（不会改动你的检出）；请先把该工作区切到其他分支`,
};
}
return { ok: false, error: `创建临时合并 worktree 失败：${msg}` };
}
try {
// 2) 合并（--no-ff：保留任务分支的合并节点）
await git(dir, [
'-c', 'user.name=maestro', '-c', 'user.email=maestro@local',
'merge', '--no-ff', branch, '-m', `merge: ${branch} [${taskId}]`,
]);
const mergeCommit = (await git(dir, ['rev-parse', 'HEAD'])).trim();
return { ok: true, mergeCommit };
} catch (e) {
// 3) 冲突/失败：收集冲突文件 → abort → 返回错误（defaultBranch 与任务分支无损）
let conflicted = '';
try {
conflicted = (await git(dir, ['diff', '--name-only', '--diff-filter=U'])).trim();
} catch { /* 收集失败不影响报错 */ }
await git(dir, ['merge', '--abort']).catch(() => undefined);
const files = conflicted ? `；冲突文件：${conflicted.split('\n').join('、')}` : '';
return { ok: false, error: `${(e as Error).message}${files}` };
} finally {
// 4) 无论成败都清理临时 worktree
await git(repoPath, ['worktree', 'remove', '--force', dir]).catch(() => undefined);
await git(repoPath, ['worktree', 'prune']).catch(() => undefined);
rmSync(dir, { recursive: true, force: true });
}
}
src/executor/reviewer.ts
+63 -18
View File
@@ -3,6 +3,9 @@ import type { WorktreeInfo } from './worktree.js';
import { runClaude } from './cc.js';
import { pickModel } from './models.js';
/** 复审角色：code（code review）/ security（安全审计）。两者共用 models.ts 的 reviewer 模型映射。 */
export type ReviewRole = 'code' | 'security';
export interface ReviewResult {
summary: string; // 最终文本去掉 VERDICT 行（markdown）
verdict: ReviewVerdict | null; // 解析不到 = null
@@ -10,7 +13,7 @@ export interface ReviewResult {
sessionId: string | null;
}
/** 复审函数签名（orchestrator 依赖注入点；测试传 mock，生产传 reviewTask）。失败时抛错，由 orchestrator 兜底。 */
/** 复审函数签名（orchestrator 依赖注入点；测试传 mock，生产传 reviewCode/reviewSecurity）。失败时抛错，由 orchestrator 兜底。 */
export type ReviewerFn = (
task: Task,
project: Project,
@@ -22,14 +25,53 @@ export type ReviewerFn = (
const REVIEW_MAX_TURNS = 40;
const REVIEW_TIMEOUT_MS = 15 * 60_000; // 15min
/** 组装复审提示词：任务说明 + 执行者自述 + 审 diff + 固定模板输出 */
export function buildReviewPrompt(task: Task, project: Project, worktree: WorktreeInfo, executorReport: string): string {
/** 各角色的开场白 + 固定输出模板 */
const ROLE_TEXT: Record<ReviewRole, { title: string; intro: string; template: string[] }> = {
code: {
title: '复审任务',
intro: '你是独立的代码复审员。一个执行 agent 刚在当前 git worktree 完成了下述任务，请你复审它的改动。',
template: [
'## 做了什么',
'## 怎么做的',
'## 测试情况（有没有测试、跑了没有、结果如何；无测试体系则说明为什么不适用）',
'## Code Review（正确性/可维护性问题，逐条带文件:行号）',
'## 结论',
'（建议通过 或 建议拒绝；拒绝必须给出理由与改进方案）',
'VERDICT: approve|reject',
],
},
security: {
title: '安全审计任务',
intro: '你是独立的安全审计员。一个执行 agent 刚在当前 git worktree 完成了下述任务，请你只从安全角度审计它的改动。',
template: [
'## 安全审计',
'### 凭证与密钥泄露（硬编码 token/密码/私钥、误提交的敏感文件）',
'### 注入与危险命令（shell/SQL/路径注入、eval、不受控的命令拼接）',
'### 权限与边界（越权读写、worktree 之外的副作用、过宽的白名单）',
'### UI 文案红线词（面向用户文案中的违规/敏感表述）',
'### 生产部署风险（影响线上服务、不可回滚操作、配置误伤）',
'## 结论',
'（建议通过 或 建议拒绝；拒绝必须给出理由与改进方案）',
'VERDICT: approve|reject',
],
},
};
/** 组装复审提示词：任务说明 + 执行者自述 + 审 diff + 按角色的固定模板输出 */
export function buildReviewPrompt(
role: ReviewRole,
task: Task,
project: Project,
worktree: WorktreeInfo,
executorReport: string,
): string {
const body = task.operations ?? task.spec ?? task.plan ?? '';
const rt = ROLE_TEXT[role];
return [
`# 复审任务：${task.title}`,
`# ${rt.title}：${task.title}`,
`任务 ID：${task.id}`,
'',
'你是独立的代码复审员。一个执行 agent 刚在当前 git worktree 完成了下述任务，请你复审它的改动。',
rt.intro,
'',
'## 任务原始要求',
body || '（无详细说明，以标题为准）',
@@ -44,14 +86,7 @@ export function buildReviewPrompt(task: Task, project: Project, worktree: Worktr
'',
'## 输出格式（最终回复必须严格按此模板，markdown）',
'```',
'## 做了什么',
'## 怎么做的',
'## 测试情况（有没有测试、跑了没有、结果如何；无测试体系则说明为什么不适用）',
'## Code Review（正确性/可维护性问题，逐条带文件:行号）',
'## 安全 Review（凭证泄露/注入/权限/危险操作；本项目还需检查 UI 文案红线词与生产部署安全）',
'## 结论',
'（建议通过 或 建议拒绝；拒绝必须给出理由与改进方案）',
'VERDICT: approve|reject',
...rt.template,
'```',
'- 最后一行必须是单独一行 `VERDICT: approve` 或 `VERDICT: reject`（机器解析用，不要带其他内容）。',
].join('\n');
@@ -77,10 +112,11 @@ export function parseVerdict(text: string): { summary: string; verdict: ReviewVe
}
/**
* 执行后自动复审：在 worktree 内再起一个只读 headless CC，审 diff 并产出固定模板 summary + verdict。
* 通用复审：在 worktree 内再起一个只读 headless CC，审 diff 并产出按角色模板的 summary + verdict。
* 转录写 <transcriptDir>/<runId>.jsonl。失败抛错（orchestrator 兜底，不挡任务）。
*/
export async function reviewTask(
export async function runReview(
role: ReviewRole,
task: Task,
project: Project,
worktree: WorktreeInfo,
@@ -89,7 +125,7 @@ export async function reviewTask(
): Promise<ReviewResult> {
const model = pickModel(task, project, 'reviewer');
const cc = await runClaude({
prompt: buildReviewPrompt(task, project, worktree, executorReport),
prompt: buildReviewPrompt(role, task, project, worktree, executorReport),
cwd: worktree.dir,
model,
runId,
@@ -103,9 +139,18 @@ export async function reviewTask(
],
});
if (!cc.ok) throw new Error(cc.error ?? '复审执行失败');
if (!cc.finalText.trim()) throw new Error('复审未产出最终文本');
const label = role === 'code' ? '复审' : '安全审计';
if (!cc.ok) throw new Error(cc.error ?? `${label}执行失败`);
if (!cc.finalText.trim()) throw new Error(`${label}未产出最终文本`);
const { summary, verdict } = parseVerdict(cc.finalText);
return { summary, verdict, transcriptRef: cc.transcriptRef, sessionId: cc.sessionId };
}
/** code review 入口（kind=reviewer 的 run） */
export const reviewCode: ReviewerFn = (task, project, worktree, runId, executorReport) =>
runReview('code', task, project, worktree, runId, executorReport);
/** 安全审计入口（kind=security 的 run） */
export const reviewSecurity: ReviewerFn = (task, project, worktree, runId, executorReport) =>
runReview('security', task, project, worktree, runId, executorReport);
src/model/types.ts
+7 -5
View File
@@ -40,9 +40,11 @@ export interface TaskResult {
worktree: string | null;
diffSummary: string | null;
commits: string[];
prUrl: string | null;
summary: string | null; // 复审 summary（markdown），旧数据缺省 null
verdict: ReviewVerdict | null; // 复审结论，解析不到 / 复审失败 = null
prUrl: string | null; // 合并后写 merged:<mergeCommit>（复用字段记录合并产物）
summary: string | null; // code review 报告（markdown），旧数据缺省 null
verdict: ReviewVerdict | null; // code review 结论，解析不到 / 复审失败 = null
securitySummary: string | null; // 安全审计报告（markdown），旧数据缺省 null
securityVerdict: ReviewVerdict | null; // 安全审计结论，解析不到 / 审计失败 = null
}
export interface Task {
@@ -66,13 +68,13 @@ export interface Task {
updatedAt: string;
}
export type RunKind = 'planner' | 'executor' | 'reviewer';
export type RunKind = 'planner' | 'executor' | 'reviewer' | 'security';
export type RunStatus = 'started' | 'succeeded' | 'failed' | 'cancelled';
export interface Run {
id: Id;
taskId: Id;
kind: RunKind; // planner（产出拆解/方案）| executor（执行改动）| reviewer（执行后自动复审）
kind: RunKind; // planner（产出拆解/方案）| executor（执行改动）| reviewer（code review）| security（安全审计）
worktree: string | null;
branch: string | null;
status: RunStatus;
src/store/mappers.ts
+5 -2
View File
@@ -39,9 +39,10 @@ export function rowToProject(r: ProjectRow): Project {
};
}
/** 旧 result JSON 兜底：summary/verdict 是后加字段，老数据缺省补 null */
/** 旧 result JSON 兜底：summary/verdict/securitySummary/securityVerdict 是后加字段，老数据缺省补 null */
function parseResult(json: string): TaskResult {
const raw = JSON.parse(json) as Partial<TaskResult>;
const verdict = (v: unknown): TaskResult['verdict'] => (v === 'approve' || v === 'reject' ? v : null);
return {
branch: raw.branch ?? null,
worktree: raw.worktree ?? null,
@@ -49,7 +50,9 @@ function parseResult(json: string): TaskResult {
commits: raw.commits ?? [],
prUrl: raw.prUrl ?? null,
summary: raw.summary ?? null,
verdict: raw.verdict === 'approve' || raw.verdict === 'reject' ? raw.verdict : null,
verdict: verdict(raw.verdict),
securitySummary: raw.securitySummary ?? null,
securityVerdict: verdict(raw.securityVerdict),
};
}
src/store/schema.sql
+1 -1
View File
@@ -53,7 +53,7 @@ CREATE INDEX IF NOT EXISTS idx_approvals_task ON approvals(task_id, at);
CREATE TABLE IF NOT EXISTS runs (
id TEXT PRIMARY KEY,
task_id TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
kind TEXT NOT NULL, -- planner | executor
kind TEXT NOT NULL, -- planner | executor | reviewer | security
worktree TEXT,
branch TEXT,
status TEXT NOT NULL, -- started | succeeded | failed | cancelled