import Fastify, { type FastifyInstance } from 'fastify'; import { WebSocketServer, type WebSocket } from 'ws'; import { Store, StoreError, type ActiveRun, type PatchProjectInput, type PatchTaskInput, type UserSettings } 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 { acceptAndMerge } from '../executor/exec-merge.js'; import { createWorktree, git } from '../executor/worktree.js'; import { resolveLogo, LOGO_MIME } from './logo.js'; import { readTranscript, TranscriptError } from '../executor/transcript.js'; import { createReadStream, createWriteStream, mkdirSync, readFileSync, rmSync, statSync } from 'node:fs'; import { transcriptDir } from '../executor/cc.js'; import { homedir } from 'node:os'; import { join, basename, resolve, sep } from 'node:path'; import { pipeline } from 'node:stream/promises'; import multipart from '@fastify/multipart'; import { createUsageFetcher, type UsageInfo } from '../daemon/usage.js'; /** 数据根目录（与 protocol/worktree 同约定）： */ function dataDir(): string { return process.env.MAESTRO_DATA_DIR ?? join(homedir(), '.maestro'); } /** 常见 MIME → 落盘扩展名（无前导点）。剪贴板粘贴 blob 常无扩展名，据此补齐。 */ const MIME_EXT: Record = { 'image/png': 'png', 'image/jpeg': 'jpg', 'image/jpg': 'jpg', 'image/gif': 'gif', 'image/webp': 'webp', 'image/svg+xml': 'svg', 'image/bmp': 'bmp', 'image/x-icon': 'ico', 'image/vnd.microsoft.icon': 'ico', 'image/tiff': 'tiff', 'image/heic': 'heic', 'image/heif': 'heif', 'image/avif': 'avif', 'application/pdf': 'pdf', 'application/zip': 'zip', 'application/json': 'json', 'application/xml': 'xml', 'text/plain': 'txt', 'text/markdown': 'md', 'text/csv': 'csv', 'text/html': 'html', }; /** MIME 类型 → 扩展名（无点）。未知映射时对 image/ 兜底取 subtype；其余返回 ''。 */ export function extFromMime(mimetype: string | undefined): string { if (!mimetype) return ''; const key = mimetype.split(';')[0].trim().toLowerCase(); if (MIME_EXT[key]) return MIME_EXT[key]; const m = key.match(/^image\/([a-z0-9.+-]+)$/); if (m) return m[1].replace(/[^a-z0-9]/g, ''); return ''; } /** basename 末尾是否带「看起来像扩展名」的后缀（1-8 位字母数字）。 */ function hasExtension(name: string): boolean { return /\.[A-Za-z0-9]{1,8}$/.test(name); } /** * 计算附件落盘安全文件名。规则： * 1. 仅取 basename，净化非法字符（`[^\w.\-]` → `_`），剥掉前导点（防隐藏/空名）。 * 2. 主名为空或为占位名（blob/image/file/unknown/untitled/paste）→ 视为「无名粘贴 blob」， * 生成唯一名 `paste--`，扩展名优先取 mimetype 推断、其次保留原扩展名。 * 3. 有正常主名但缺扩展名 → 按 mimetype 补扩展名（mimetype 未知则保持原样）。 * 返回值保证非空。 */ export function safeAttachmentName(rawName: string | undefined, mimetype: string | undefined, seq = 0): string { const ext = extFromMime(mimetype); const sanitized = basename((rawName ?? '').trim()).replace(/[^\w.\-]/g, '_').replace(/^\.+/, ''); const withExt = hasExtension(sanitized); const stem = withExt ? sanitized.replace(/\.[A-Za-z0-9]{1,8}$/, '') : sanitized; const isPlaceholder = stem === '' || /^(blob|image|file|unknown|untitled|paste)$/i.test(stem); if (isPlaceholder) { const origExt = withExt ? sanitized.slice(sanitized.lastIndexOf('.') + 1) : ''; const useExt = ext || origExt; const base = `paste-${Date.now()}-${seq}`; return useExt ? `${base}.${useExt}` : base; } if (!withExt && ext) return `${sanitized}.${ext}`; return sanitized; } /** Project 出参：附加 hasTodoJson（/todo/todo.json 是否存在，每次序列化时算） */ function projectOut(p: Project, store: Store): Project & { hasTodoJson: boolean; summary: ReturnType } { return { ...p, hasTodoJson: hasTodoJson(p.repoPath), summary: store.projectSummary(p.id) }; } 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 }); // 附件上传：multipart/form-data（单文件 ≤25MB，单次 ≤10 个） app.register(multipart, { limits: { fileSize: 25 * 1024 * 1024, files: 10 } }); // 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((p) => projectOut(p, store))); app.post('/api/projects', (req) => { const b = req.body as Record; if (!b?.name || !b?.repoPath) throw new StoreError('name 与 repoPath 必填'); // 用户级全局默认（核心策略子集）：body 显式给则用 body，否则套默认 const s = store.getSettings(); const pick = (key: string, fallback: T): T => (b[key] !== undefined ? (b[key] as T) : fallback); const created = 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: pick('autonomy', s.autonomy) as never, model: b.model !== undefined ? (b.model === null ? null : String(b.model)) : s.model, concurrency: Number(pick('concurrency', s.concurrency)), maxRetries: Number(pick('maxRetries', s.maxRetries)), timeoutMs: Number(pick('timeoutMs', s.timeoutMs)), }); // createProject 不接受的默认字段（自动放行 / 预算）经 patchProject 套用 return projectOut(store.patchProject(created.id, { autoApprovePlan: Boolean(pick('autoApprovePlan', s.autoApprovePlan)), autoApproveExec: Boolean(pick('autoApproveExec', s.autoApproveExec)), budgetUsd: ((): number | null => { const v = pick('budgetUsd', s.budgetUsd); return v === null ? null : Number(v); })(), budgetPeriod: pick('budgetPeriod', s.budgetPeriod) as 'day' | 'month', }), store); }); 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); }); // 部分更新项目配置（校验在 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); if (b.budgetUsd !== undefined) patch.budgetUsd = b.budgetUsd === null || b.budgetUsd === '' ? null : Number(b.budgetUsd); if (b.budgetPeriod !== undefined) patch.budgetPeriod = b.budgetPeriod as 'day' | 'month'; if (b.autoApprovePlan !== undefined) patch.autoApprovePlan = Boolean(b.autoApprovePlan); if (b.autoApproveExec !== undefined) patch.autoApproveExec = Boolean(b.autoApproveExec); if (b.checks !== undefined) patch.checks = b.checks === null || b.checks === '' ? null : String(b.checks); if (b.agentRules !== undefined) patch.agentRules = b.agentRules === null || b.agentRules === '' ? null : String(b.agentRules); if (b.models !== undefined) patch.models = b.models === null || b.models === '' ? null : (b.models as PatchProjectInput['models']); return projectOut(store.patchProject(id, patch), store); }); // 项目重排序（侧栏拖动）：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((p) => projectOut(p, store)); }); // ---------- 用户级全局默认配置（per-user，新建项目默认值来源）---------- app.get('/api/settings', () => store.getSettings()); app.put('/api/settings', (req) => { const b = (req.body ?? {}) as Record; const patch: Partial = {}; if (b.autonomy !== undefined) patch.autonomy = b.autonomy as Autonomy; if (b.concurrency !== undefined) patch.concurrency = Number(b.concurrency); if (b.maxRetries !== undefined) patch.maxRetries = Number(b.maxRetries); if (b.timeoutMs !== undefined) patch.timeoutMs = Number(b.timeoutMs); if (b.autoApprovePlan !== undefined) patch.autoApprovePlan = Boolean(b.autoApprovePlan); if (b.autoApproveExec !== undefined) patch.autoApproveExec = Boolean(b.autoApproveExec); if (b.budgetUsd !== undefined) patch.budgetUsd = b.budgetUsd === null || b.budgetUsd === '' ? null : Number(b.budgetUsd); if (b.budgetPeriod !== undefined) patch.budgetPeriod = b.budgetPeriod as 'day' | 'month'; if (b.model !== undefined) patch.model = b.model === null || b.model === '' ? null : String(b.model); if (b.globalConcurrency !== undefined) patch.globalConcurrency = b.globalConcurrency === null || b.globalConcurrency === '' ? 0 : Number(b.globalConcurrency); return store.putSettings(patch); }); // 彻底删除项目（连带 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 订阅额度（5h+周配额，查询失败→null）+ 成本明细（按项目/模型/总计，按 period 窗口）。 // ?period=day|month（默认 month）；?project= 限定单项目成本。 app.get('/api/usage', async (req) => { const q = req.query as { period?: string; project?: string }; const period = q.period === 'day' ? 'day' : q.period === 'week' ? 'week' : 'month'; const quota = await getUsage(); const cost = store.costSummary(period, q.project && q.project.trim() ? q.project.trim() : undefined); return { ...(quota ?? {}), cost }; }); // 用量详情：时间序列（按天/按月）+ 分项目下钻，供侧栏「详情」弹框。 app.get('/api/usage/detail', (req) => { const q = req.query as { granularity?: string }; const granularity = q.granularity === 'month' ? 'month' : q.granularity === 'week' ? 'week' : 'day'; return store.usageDetail(granularity); }); // 健康检查：daemon 存活 + DB 可读 + 在途 run 数。 app.get('/api/health', () => { let db = true; let inflight = 0; try { inflight = store.activeRuns().length; } catch { db = false; } return { ok: db, db, inflight, at: new Date().toISOString() }; }); 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, scopeFiles: Array.isArray(b.scopeFiles) ? (b.scopeFiles 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; }); // 附件上传（multipart files[]）：落盘 /tasks//attachments/，元数据写 tasks.attachments app.post('/api/tasks/:id/attachments', async (req, reply) => { const { id } = req.params as { id: string }; if (!store.getTask(id)) return reply.code(404).send({ error: `任务不存在: ${id}` }); const dir = join(dataDir(), 'tasks', id, 'attachments'); mkdirSync(dir, { recursive: true }); const saved: Array<{ name: string; type: string; path: string }> = []; let seq = 0; for await (const part of req.files()) { // 无名/无扩展名（如剪贴板粘贴 blob）→ 按 mimetype 生成安全文件名+扩展名 const safe = safeAttachmentName(part.filename, part.mimetype, seq++); await pipeline(part.file, createWriteStream(join(dir, safe))); if (part.file.truncated) return reply.code(400).send({ error: `文件 ${part.filename || safe} 超过 25MB 上限` }); saved.push({ name: part.filename || safe, type: part.mimetype, path: `tasks/${id}/attachments/${safe}` }); } if (saved.length === 0) return reply.code(400).send({ error: '未收到文件' }); return { attachments: store.addAttachments(id, saved) }; }); // 附件读取/预览/下载：默认 inline（图片可 预览）；?download=1 触发下载。 // 标识符为磁盘文件名 :name（= basename(att.path)），只服务元数据里登记过的附件。 app.get('/api/tasks/:id/attachments/:name', (req, reply) => { const { id, name } = req.params as { id: string; name: string }; const task = store.getTask(id); if (!task) return reply.code(404).send({ error: `任务不存在: ${id}` }); const att = (task.attachments ?? []).find((a) => basename(a.path) === name); if (!att) return reply.code(404).send({ error: `附件不存在: ${name}` }); const abs = resolve(join(dataDir(), att.path)); // 纵深防御：解析后必须仍在该任务 attachments 目录内 const baseDir = resolve(join(dataDir(), 'tasks', id, 'attachments')); if (abs !== baseDir && !abs.startsWith(baseDir + sep)) return reply.code(400).send({ error: '附件路径越界' }); let st; try { st = statSync(abs); } catch { return reply.code(404).send({ error: '附件文件已丢失' }); } const dl = (req.query as { download?: string }).download; const fallbackName = att.name || name; // 安全：上传文件与本 API 同源，inline 渲染会在 console 源里执行其脚本 → 存储型 XSS。 // 故只对【硬白名单图片类型】允许 inline 预览；其余（尤其 image/svg+xml、text/html 等可携带脚本的类型） // 一律强制 attachment 下载 + content-type 改 octet-stream，并加 nosniff + CSP sandbox 双重兜底。 const SAFE_INLINE = new Set(['image/png', 'image/jpeg', 'image/gif', 'image/webp']); const inline = !dl && SAFE_INLINE.has(att.type ?? ''); reply .header('cache-control', 'private, max-age=300') .header('content-length', String(st.size)) .header('x-content-type-options', 'nosniff') .header('content-security-policy', "sandbox; default-src 'none'") .header('content-disposition', `${inline ? 'inline' : 'attachment'}; filename*=UTF-8''${encodeURIComponent(fallbackName)}`) .type(inline ? (att.type as string) : 'application/octet-stream'); return reply.send(createReadStream(abs)); }); // 删除单个附件：先删元数据（拿到 removed），再尽力删磁盘文件 app.delete('/api/tasks/:id/attachments/:name', (req, reply) => { const { id, name } = req.params as { id: string; name: string }; if (!store.getTask(id)) return reply.code(404).send({ error: `任务不存在: ${id}` }); const { attachments, removed } = store.removeAttachment(id, name); if (!removed) return reply.code(404).send({ error: `附件不存在: ${name}` }); const abs = resolve(join(dataDir(), removed.path)); const baseDir = resolve(join(dataDir(), 'tasks', id, 'attachments')); if (abs === baseDir || abs.startsWith(baseDir + sep)) { try { rmSync(abs, { force: true }); } catch { /* 磁盘文件已不在，忽略 */ } } return { attachments }; }); // 人工接管：准备（或复用）任务 worktree，返回用户在自己终端起交互式 claude 的命令。 // 适用卡住/需人工的任务——人接手手动改。不在 daemon 内起交互会话（无 TTY）。 app.post('/api/tasks/:id/takeover', async (req, reply) => { const { id } = req.params as { id: string }; const task = store.getTask(id); if (!task) return reply.code(404).send({ error: `任务不存在: ${id}` }); if (store.hasInflightRun(id)) return reply.code(400).send({ error: '任务有在途 run，无法接管' }); const project = store.getProject(task.projectId); if (!project) return reply.code(404).send({ error: `项目不存在: ${task.projectId}` }); let worktree = task.result?.worktree ?? null; let branch = task.result?.branch ?? null; if (!worktree) { const wt = await createWorktree(project.repoPath, id, project.defaultBranch); worktree = wt.dir; branch = wt.branch; } return { taskId: id, worktree, branch, console: `cd ${worktree} && claude` }; }); // 部分更新任务（title/priority/complexity；complexity 重置逻辑在 Store.patchTask） app.patch('/api/tasks/:id', (req) => { const { id } = req.params as { id: string }; if (store.hasInflightRun(id)) { throw new StoreError('任务有在途 run，执行期间禁止修改'); } 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[]; } if (b.scopeFiles !== undefined) { if (b.scopeFiles !== null && (!Array.isArray(b.scopeFiles) || !b.scopeFiles.every((f) => typeof f === 'string'))) { throw new StoreError('scopeFiles 必须是文件 glob 字符串数组或 null'); } patch.scopeFiles = b.scopeFiles as string[] | null; } 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); }); // 任务改动 diff（结果评审用）：git diff defaultBranch...branch，按文件切分返回，供点击文件看具体改动 app.get('/api/tasks/:id/diff', async (req) => { const { id } = req.params as { id: string }; const task = store.getTask(id); if (!task) throw new StoreError(`任务不存在: ${id}`); const project = store.getProject(task.projectId); const branch = task.result?.branch; if (!project || !branch) return { files: [] }; let raw = ''; try { raw = await git(project.repoPath, ['diff', `${project.defaultBranch}...${branch}`]); } catch { return { files: [] }; } const files = raw.split(/^diff --git /m).filter(Boolean).map((p) => { const m = p.match(/^a\/(\S+)\s+b\/(\S+)/); const path = m ? m[2] : (p.split('\n')[0] || '').trim(); return { path, diff: 'diff --git ' + p }; }); return { files }; }); // 实时流（SSE）：跟随任务「最近一条 run」的 transcript 文件，把新写入的 agent 消息逐行推送。 // 活跃 run（status=started）= 真实时；已结束 run = 一次性回放后 end。客户端 EventSource 消费。 app.get('/api/tasks/:id/stream', (req, reply) => { const { id } = req.params as { id: string }; const runs = store.listRuns(id); const run = runs.find((r) => r.status === 'started') ?? runs[runs.length - 1]; if (!run) return reply.code(404).send({ error: '该任务暂无 run 可流式' }); const file = join(transcriptDir(), `${run.id}.jsonl`); reply.hijack(); reply.raw.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', Connection: 'keep-alive', 'X-Accel-Buffering': 'no', }); reply.raw.write(`event: open\ndata: {"runId":"${run.id}","kind":"${run.kind}"}\n\n`); let offset = 0; let closed = false; const stop = () => { if (closed) return; closed = true; clearInterval(timer); reply.raw.end(); }; const tick = (): void => { if (closed) return; try { const size = statSync(file).size; if (size > offset) { const chunk = readFileSync(file).subarray(offset).toString('utf8'); offset = size; for (const line of chunk.split('\n')) if (line.trim()) reply.raw.write(`data: ${line}\n\n`); } } catch { /* 文件尚未创建：等下一拍 */ } const cur = store.getRun(run.id); if (!cur || cur.status !== 'started') { reply.raw.write('event: end\ndata: {}\n\n'); stop(); } }; const timer = setInterval(tick, 500); tick(); req.raw.on('close', stop); }); // 执行转录回放：读 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 outcome = await acceptAndMerge(store, app.log, task, b.actor ?? 'user'); if (!outcome.ok) { if (outcome.remediationTaskId) { const rem = store.getTask(outcome.remediationTaskId); throw new StoreError(`合并失败：${outcome.error}（原任务保留在审核闸；已建最高优先级补救任务 ${rem?.id}「${rem?.title}」来完成合并）`); } throw new StoreError(`合并失败：${outcome.error}`); } 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; }