import Fastify, { type FastifyInstance } from 'fastify'; import { WebSocketServer, type WebSocket } from 'ws'; import { Store, StoreError, type ActiveRun, type PatchProjectInput, type PatchTaskInput } from '../store/index.js'; import type { Complexity } from '../model/complexity.js'; import { isComplexity } from '../model/complexity.js'; 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 { classifyComplexity, type ClassifierFn } from '../executor/classify.js'; import { mergeBranch } from '../executor/merge.js'; import { git, removeWorktree } from '../executor/worktree.js'; import { resolveLogo, LOGO_MIME } from './logo.js'; import { readTranscript, TranscriptError } from '../executor/transcript.js'; import { createReadStream } from 'node:fs'; import { createUsageFetcher, type UsageInfo } from '../daemon/usage.js'; /** Project 出参:附加 hasTodoJson(/todo/todo.json 是否存在,每次序列化时算) */ function projectOut(p: Project): Project & { hasTodoJson: boolean } { return { ...p, hasTodoJson: hasTodoJson(p.repoPath) }; } export interface ApiOptions { store: Store; logger?: boolean; /** Claude 订阅额度查询(测试注入;默认直连 OAuth usage API,60s 缓存) */ getUsage?: () => Promise; /** AUTO 复杂度分类器(测试注入;默认 classifyComplexity,用 sonnet 起一次轻量调用) */ classify?: ClassifierFn; } /** * REST + WebSocket API。所有写操作走 Store(带状态机守卫 / 审批闸)。 * Store 的事件经 subscribe 广播到所有 WS 客户端,驱动看板实时刷新。 */ export function buildServer(opts: ApiOptions): FastifyInstance { const { store } = opts; const classify = opts.classify ?? ((title, description, cwd) => classifyComplexity(title, description, { cwd })); const app = Fastify({ logger: opts.logger ?? false }); // Store 错误 → 400(业务校验),其余 → 500 app.setErrorHandler((err, _req, reply) => { if (err instanceof StoreError) return reply.code(400).send({ error: err.message }); app.log.error(err); return reply.code(500).send({ error: err.message ?? 'internal error' }); }); // ---------- Projects ---------- app.get('/api/projects', () => store.listProjects().map(projectOut)); app.post('/api/projects', (req) => { const b = req.body as Record; if (!b?.name || !b?.repoPath) throw new StoreError('name 与 repoPath 必填'); return projectOut(store.createProject({ name: String(b.name), repoPath: String(b.repoPath), defaultBranch: b.defaultBranch ? String(b.defaultBranch) : undefined, verifyCmd: b.verifyCmd === undefined ? undefined : (b.verifyCmd === null ? null : String(b.verifyCmd)), autonomy: b.autonomy as never, model: b.model === undefined ? undefined : (b.model === null ? null : String(b.model)), concurrency: b.concurrency === undefined ? undefined : Number(b.concurrency), maxRetries: b.maxRetries === undefined ? undefined : Number(b.maxRetries), timeoutMs: b.timeoutMs === undefined ? undefined : Number(b.timeoutMs), })); }); app.get('/api/projects/:id', (req) => { const { id } = req.params as { id: string }; const p = store.getProject(id); if (!p) throw new StoreError(`项目不存在: ${id}`); return projectOut(p); }); // 部分更新项目配置(校验在 Store.patchProject) app.patch('/api/projects/:id', (req) => { const { id } = req.params as { id: string }; const b = (req.body ?? {}) as Record; const patch: PatchProjectInput = {}; if (b.autonomy !== undefined) patch.autonomy = b.autonomy as Autonomy; if (b.concurrency !== undefined) patch.concurrency = Number(b.concurrency); if (b.status !== undefined) patch.status = b.status as 'active' | 'paused'; if (b.verifyCmd !== undefined) patch.verifyCmd = b.verifyCmd === null ? null : String(b.verifyCmd); if (b.model !== undefined) patch.model = b.model === null ? null : String(b.model); if (b.logo !== undefined) patch.logo = b.logo === null || b.logo === '' ? null : String(b.logo); if (b.maxRetries !== undefined) patch.maxRetries = Number(b.maxRetries); if (b.timeoutMs !== undefined) patch.timeoutMs = Number(b.timeoutMs); return projectOut(store.patchProject(id, patch)); }); // 项目重排序(侧栏拖动):body { order: [projectId, ...] } app.post('/api/projects/reorder', (req) => { const b = (req.body ?? {}) as { order?: unknown }; if (!Array.isArray(b.order)) throw new StoreError('order 必须是项目 id 数组'); return store.reorderProjects(b.order.map(String)).map(projectOut); }); // 彻底删除项目(连带 tasks/runs/approvals/events,不可恢复) app.delete('/api/projects/:id', (req) => { const { id } = req.params as { id: string }; store.deleteProject(id); return { ok: true }; }); // 项目 logo:仓库内文件 → 流式返回;外链头像 → 302;无 → 404(前端用首字母徽章兜底) app.get('/api/projects/:id/logo', async (req, reply) => { const { id } = req.params as { id: string }; const p = store.getProject(id); if (!p) return reply.code(404).send(); const r = await resolveLogo(p); if (!r) return reply.code(404).send(); if (r.type === 'redirect') return reply.redirect(r.url); const ext = r.path.split('.').pop()?.toLowerCase() ?? ''; reply.header('cache-control', 'no-cache').type(LOGO_MIME[ext] ?? 'application/octet-stream'); return reply.send(createReadStream(r.path)); }); // 单向同步 /todo/todo.json → maestro(导入=首次同步,幂等) app.post('/api/projects/:id/sync', (req) => { const { id } = req.params as { id: string }; return syncProject(store, id); }); // ---------- Agents(每项目一条;active 来自 runs 表 status='started',执行器 Phase 2 前通常为空) ---------- // 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(); for (const r of runs) { const list = byProject.get(r.projectId) ?? []; list.push(r); byProject.set(r.projectId, list); } const agents = store.listProjects().map((p) => ({ projectId: p.id, projectName: p.name, autonomy: p.autonomy, concurrency: p.concurrency, status: p.status, scheduling: 'score', // 调度模式:score 优先(自身分+链条惯性+解锁加权) models: resolvedExecutorModels(p), // 各复杂度实际使用的执行模型 active: (byProject.get(p.id) ?? []).map((r) => ({ runId: r.runId, taskId: r.taskId, taskTitle: r.taskTitle, kind: r.kind, startedAt: r.startedAt, })), })); return { totalActive: runs.length, agents, usage: await getUsage() }; }); // Claude 订阅额度(5 小时 + 周窗口)单独透传;查询失败 → null(前端/CLI 降级显示) app.get('/api/usage', async () => await getUsage()); app.get('/api/projects/:id/tasks', (req) => { const { id } = req.params as { id: string }; return store.listTasks(id); }); app.get('/api/projects/:id/events', (req) => { const { id } = req.params as { id: string }; return store.listEvents(id); }); app.get('/api/projects/:id/next', (req) => { const { id } = req.params as { id: string }; return store.nextExecutable(id) ?? { next: null }; }); // ---------- Metrics(健康度 + 成本聚合)---------- // GET /api/metrics?project=;不带 project = 全部项目聚合。 app.get('/api/metrics', (req) => { const q = req.query as { project?: string }; return store.metrics(q.project && q.project.trim() ? q.project.trim() : undefined); }); // ---------- Tasks ---------- app.post('/api/projects/:id/tasks', (req) => { const { id } = req.params as { id: string }; const b = req.body as Record; if (!b?.title) throw new StoreError('title 必填'); // 'auto' = 由模型判定:先以 medium 落库占位,建任务后异步回填(不阻塞返回) const auto = b.complexity === 'auto'; if (!auto && !isComplexity(b.complexity)) throw new StoreError('complexity 必须是 auto|hard|medium|easy'); const task = store.createTask({ projectId: id, title: String(b.title), complexity: auto ? 'medium' : (b.complexity as Complexity), parentId: b.parentId ? String(b.parentId) : null, priority: b.priority === undefined ? undefined : Number(b.priority), deps: Array.isArray(b.deps) ? (b.deps as string[]) : undefined, }); if (auto) { // 异步分类回填:classifyComplexity 自身永不抛错(失败兜底 medium),外层再兜一层防御 const cwd = store.getProject(id)?.repoPath ?? process.cwd(); void classify(task.title, undefined, cwd) .then(({ complexity, reason }) => store.applyAutoComplexity(task.id, complexity, reason)) .catch((e: unknown) => app.log.error(`任务 ${task.id} AUTO 复杂度分类失败(保留占位 medium):${(e as Error).message}`)); } return task; }); app.get('/api/tasks/:id', (req) => { const { id } = req.params as { id: string }; const t = store.getTask(id); if (!t) throw new StoreError(`任务不存在: ${id}`); return t; }); // 部分更新任务(title/priority/complexity;complexity 重置逻辑在 Store.patchTask) app.patch('/api/tasks/:id', (req) => { const { id } = req.params as { id: string }; const b = (req.body ?? {}) as Record; const patch: PatchTaskInput = {}; if (b.title !== undefined) patch.title = String(b.title); if (b.priority !== undefined) { const n = Number(b.priority); if (!Number.isFinite(n)) throw new StoreError('priority 必须是数字'); patch.priority = n; } if (b.complexity !== undefined) { if (!isComplexity(b.complexity)) throw new StoreError('complexity 必须是 hard|medium|easy'); patch.complexity = b.complexity; } if (b.deps !== undefined) { if (!Array.isArray(b.deps) || !b.deps.every((d) => typeof d === 'string')) { throw new StoreError('deps 必须是任务 id 字符串数组'); } patch.deps = b.deps as string[]; } return store.patchTask(id, patch); }); app.get('/api/tasks/:id/children', (req) => { const { id } = req.params as { id: string }; return store.childrenOf(id); }); app.get('/api/tasks/:id/runs', (req) => { const { id } = req.params as { id: string }; return store.listRuns(id); }); // 执行转录回放:读 run.transcriptRef 指向的 jsonl,逐行解析成脱敏后的结构化消息数组。 // 校验:runId 合法 → 关联任务/项目存在(可访问)→ 有转录记录 → 路径在 transcriptDir 下(防穿越)。 app.get('/api/runs/:id/transcript', (req, reply) => { const { id } = req.params as { id: string }; const run = store.getRun(id); if (!run) return reply.code(404).send({ error: `run 不存在: ${id}` }); const task = store.getTask(run.taskId); if (!task) return reply.code(404).send({ error: `run 关联任务不存在: ${run.taskId}` }); if (!store.getProject(task.projectId)) return reply.code(404).send({ error: `run 关联项目不存在: ${task.projectId}` }); if (!run.transcriptRef) return reply.code(404).send({ error: '该 run 没有转录记录' }); try { const messages = readTranscript(run.transcriptRef); return { runId: run.id, taskId: run.taskId, kind: run.kind, status: run.status, messages }; } catch (e) { if (e instanceof TranscriptError) { return reply.code(e.code === 'invalid_path' ? 400 : 404).send({ error: e.message }); } throw e; } }); // 单任务全量事件(升序):归档详情的状态流转时间线 app.get('/api/tasks/:id/events', (req) => { const { id } = req.params as { id: string }; return store.listTaskEvents(id); }); app.post('/api/tasks/:id/plan', (req) => { const { id } = req.params as { id: string }; const b = req.body as { plan?: string }; if (!b?.plan) throw new StoreError('plan 必填'); return store.setPlan(id, b.plan); }); app.post('/api/tasks/:id/spec', (req) => { const { id } = req.params as { id: string }; const b = req.body as { spec?: string }; if (!b?.spec) throw new StoreError('spec 必填'); return store.setSpec(id, b.spec); }); app.post('/api/tasks/:id/operations', (req) => { const { id } = req.params as { id: string }; const b = req.body as { operations?: string }; if (!b?.operations) throw new StoreError('operations 必填'); return store.setOperations(id, b.operations); }); app.post('/api/tasks/:id/transition', (req) => { const { id } = req.params as { id: string }; const b = req.body as { to?: string; meta?: Record }; if (!b?.to) throw new StoreError('to 必填'); return store.transition(id, b.to as TaskStatus, b.meta ?? {}); }); // 取消任务(transition → cancelled;状态机已守卫合法流转) app.post('/api/tasks/:id/cancel', (req) => { const { id } = req.params as { id: string }; return store.transition(id, 'cancelled', { by: 'user' }); }); // 彻底删除任务及其子孙(不可恢复,cascade) app.delete('/api/tasks/:id', (req) => { const { id } = req.params as { id: string }; return store.deleteTask(id); }); // needs_attention 任务一键重投:重置重试基线 + 转 queued,让编排器下一轮重新领取 app.post('/api/tasks/:id/requeue', (req) => { const { id } = req.params as { id: string }; return store.requeueTask(id); }); // 审批闸:accept / reject(reject 必带 reason) // 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; 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) { // 自动合并失败(多为冲突)→ 建/复用最高优先级补救任务来完成合并,原任务留在审核闸 const rem = store.ensureMergeRemediationTask(task.id, { branch, targetBranch: project.defaultBranch, conflictError: mr.error ?? '未知冲突', conflictFiles: mr.conflictFiles, }); throw new StoreError(`合并失败:${mr.error}(原任务保留在审核闸;已建最高优先级补救任务 ${rem.id}「${rem.title}」来完成合并)`); } store.decide(id, 'accept', b.actor ?? 'user', b.reason ?? null); // 合并产物记录:复用 prUrl 字段写 merged: store.setResult(id, { ...task.result, prUrl: `merged:${mr.mergeCommit}` }); // 补救完成自动收口:若本任务是某原任务的「合并冲突补救」任务,合并成功后把原任务也标 done const origin = store.findMergeOriginTask(id); if (origin) { try { store.remediateOrigin(origin.id, id, mr.mergeCommit); } catch (e) { app.log.error(`补救任务 ${id} 合并后收口原任务 ${origin.id} 失败(不影响本任务结果):${(e as Error).message}`); } } // 异步回收:执行 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); }); // ---------- Approvals ---------- app.get('/api/approvals', (req) => { const q = req.query as { projectId?: string }; return store.pendingApprovals(q.projectId); }); return app; } /** 在 Fastify 的底层 http server 上挂 WebSocket,把 Store 事件广播给所有客户端。 */ export function attachWebSocket(app: FastifyInstance, store: Store): WebSocketServer { const wss = new WebSocketServer({ server: app.server, path: '/ws' }); const clients = new Set(); wss.on('connection', (ws) => { clients.add(ws); ws.on('close', () => clients.delete(ws)); ws.on('error', () => clients.delete(ws)); }); store.subscribe((evt) => { const msg = JSON.stringify(evt); for (const ws of clients) { if (ws.readyState === ws.OPEN) ws.send(msg); } }); return wss; }