/** * Graph Run Routes — P3 图执行的 REST 端点(T6) * * 挂载: /api/graph-runs(主 router) * /api/company(companyRouter,只提供 GET /:companyId/graph-runs) * * 设计见 docs/todo/system-design/27-p3-p4-work-graph-console.md * §四.C 五个控制操作的可用条件与语义(本文件只做「条件判断 + 转调引擎」,不含推进逻辑) * §三.2 两处不能照抄 pipeline 的坑: * ① 定义层与运行时层分开 → GET /:id 吐的是 **run 实例**(run + nodes + edges + 回退记录), * 不是图定义,前端不会再出现 CompanyDashboard.vue 那种「拿定义画运行时高亮」的空引用; * ② 「重试」必须是独立、参数自解释的端点 → POST /:id/nodes/:nodeId/retry, * 要重试哪个节点全写在 URL 里,不需要调用方拼隐藏参数组合 * (pipeline 的 retryPipelineRun 调 /advance 但不传 stepIndex,点了必定 400)。 * * 错误约定(与 routes/company.js 的 org-graph 端点、routes/devloop.js 保持一致): * 404 { error } run / 节点不存在 * 400 { error, op, runStatus, allowed } 操作在当前 run_status 下不可用(非法状态转换,明确说清) * 400 { error, scope, from, to } 引擎抛 InvalidTransitionError * 400 { error } 入参不合法 * 500 { error } 其它 */ const express = require('express'); const engine = require('../services/graph-engine'); const { getRun, getRunState, getRunNode, getRunNodeByDefId, listChildRuns, listCompanyRuns, listLoopRecords, edgeContracts, findFiles, InvalidTransitionError, } = require('../services/work-graph-db'); const { companyRootDir } = require('../services/company-paths'); const router = express.Router(); const companyRouter = express.Router(); // ──────────────────────────────────────────────────────────────── // 小工具 // ──────────────────────────────────────────────────────────────── /** async handler 包装:抛出来的异常统一落到下面的 sendError */ const h = (fn) => (req, res) => Promise.resolve(fn(req, res)).catch((err) => sendError(res, err)); function sendError(res, err) { if (err instanceof InvalidTransitionError || err.name === 'InvalidTransitionError') { return res.status(400).json({ error: err.message, scope: err.scope, from: err.from, to: err.to }); } res.status(500).json({ error: err.message }); } /** 取 run,不存在直接回 404 并返回 null(调用方 return 掉即可) */ function loadRun(req, res) { const run = getRun(req.params.id); if (!run) { res.status(404).json({ error: `graph run 不存在: ${req.params.id}` }); return null; } return run; } /** * §四.C 的「可用条件」那一列。不可用时返回明确错误——说清是哪个操作、 * 当前是什么 run_status、什么状态下才可用,不闷掉。 * * @param {boolean} [alsoOk] 状态之外的额外放行条件(「启动」用,见下); * allowed 只用于报错文案,所以额外条件也要在里面写清楚。 */ function ensureStatus(res, run, op, allowed, alsoOk = false) { if (alsoOk || allowed.includes(run.run_status)) return true; res.status(400).json({ error: `「${op}」在 run_status='${run.run_status}' 下不可用,仅当 run_status ∈ {${allowed.join(', ')}} 时可用`, op, runStatus: run.run_status, allowed, }); return false; } /** * advanceOne 的 ok:false 分两类: * - 「操作本身被状态挡住」→ 400,用户需要先改状态(先启动 / 先恢复); * - 「这一步没推进但不算错」(没有 ready 节点 = 可能已收敛、被别人抢走、在等子 run)→ 200 原样返回。 */ const BLOCKED_REASONS = { 'run-not-found': '该 run 不存在', 'run-not-running': 'run 不在 running 状态,先调 /start', 'pause-requested': 'run 已请求暂停,先调 /start 恢复', 'stop-requested': 'run 已请求停止,不能再推进', }; function sendAdvance(res, out) { if (!out.ok && BLOCKED_REASONS[out.reason]) { return res.status(400).json({ error: BLOCKED_REASONS[out.reason], op: 'next', reason: out.reason, runStatus: out.runStatus, }); } res.json(out); } // ──────────────────────────────────────────────────────────────── // 控制类端点(§四.C 五个操作) // ──────────────────────────────────────────────────────────────── /** * POST /api/graph-runs/:id/start — 「启动」 * body: { mode?: 'manual' | 'auto' }(不传沿用 run 上已存的 execution_mode) * 可用条件:run_status ∈ {ready, paused, blocked},**或 running 且 pause_requested=1** * * 后半条不是可选的宽松(R6-F3 问题 1):「暂停」是软信号,running → paused 要等节点边界 * (landPause 手头有 running 节点就不落)。而 org 层每个节点都是 subgraph,子 run 在飞时 * 父节点恒为 running —— root run 于是**永久停在 `running` + `pause_requested=1`**, * 只看 run_status 的话「启动」永远 400,用户只能一步步点「下一步」,验收 4 的后半不成立。 * 引擎侧本来就对:startRun → startRunSync 会清全树软信号并把级联 paused 的后代拉回 running, * 这里放行的正是「撤回暂停意图」。已经在 stop_requested 的 run 不在此列(它正在收摊)。 * * auto 模式**不等**引擎跑完就返回:一次自动跑到底是分钟~小时级的, * 绝不能像 DevLoop 那样把整个循环塞进一次 HTTP 请求-响应周期(§三.1)。 */ router.post('/:id/start', h((req, res) => { const run = loadRun(req, res); if (!run) return; const pausing = run.run_status === 'running' && Boolean(run.pause_requested) && !run.stop_requested; if (!ensureStatus(res, run, '启动', ['ready', 'paused', 'blocked', 'running(pause_requested=1)'], pausing)) return; const { mode } = req.body || {}; if (mode !== undefined && mode !== 'manual' && mode !== 'auto') { return res.status(400).json({ error: `mode 只能是 'manual' 或 'auto',收到: ${mode}` }); } // driver 是 auto 循环的 promise,不可序列化也不该 await —— 丢掉,让它自己在后台跑 const { driver, ...out } = engine.startRun(run.id, mode ? { mode } : {}); res.json({ ...out, executionMode: mode || run.execution_mode, auto: Boolean(driver) }); })); /** * POST /api/graph-runs/:id/next — 「下一步」 * body: { task?: string }(可选,透传给节点 executor 作任务描述) * 可用条件:run_status ∈ {running, paused} * * 只推进**一个** ready 节点,完成后把满足条件的下游置 ready 就返回,不继续往下跑。 * 会真正发起一次 agent 调用,所以这个请求本身可能是秒~分钟级的(§四.C 表格最后一列)。 */ router.post('/:id/next', h(async (req, res) => { const run = loadRun(req, res); if (!run) return; if (!ensureStatus(res, run, '下一步', ['running', 'paused'])) return; const out = await engine.nextStep(run.id, req.body && req.body.task ? { task: req.body.task } : {}); sendAdvance(res, out); })); /** * POST /api/graph-runs/:id/pause — 「暂停」 * 可用条件:run_status = 'running' * 软信号:不打断正在 running 的节点,在下一个节点边界生效。 */ router.post('/:id/pause', h((req, res) => { const run = loadRun(req, res); if (!run) return; if (!ensureStatus(res, run, '暂停', ['running'])) return; res.json({ ...engine.pauseRun(run.id), runStatus: (getRun(run.id) || {}).run_status }); })); /** * POST /api/graph-runs/:id/stop — 「停止」 * 可用条件:run_status ∈ {running, paused, blocked} * 仍 pending/ready 的节点批量 skipped,run → stopped。正在 running 的节点能否硬取消见 T9。 */ router.post('/:id/stop', h((req, res) => { const run = loadRun(req, res); if (!run) return; if (!ensureStatus(res, run, '停止', ['running', 'paused', 'blocked'])) return; res.json(engine.stopRun(run.id)); })); /** * POST /api/graph-runs/:id/nodes/:nodeId/retry — 「重试」 * 可用条件:节点 status ∈ {fail, blocked} * * §三.2 坑 2 的正解:重试的对象写在 URL 里,端点参数自解释,body 可以是空的。 * nodeId 优先按 graph_run_nodes.id 解析;解析不到再按 def_node_id 在本 run 内找一次 * ——画布上拿到的可能是定义层节点 id,两种都能直接重试,不用调用方自己换算。 */ router.post('/:id/nodes/:nodeId/retry', h((req, res) => { const run = loadRun(req, res); if (!run) return; const { nodeId } = req.params; let nodeRun = getRunNode(nodeId); if (!nodeRun || nodeRun.run_id !== run.id) nodeRun = getRunNodeByDefId(run.id, nodeId); if (!nodeRun) { return res.status(404).json({ error: `节点 ${nodeId} 不属于 run ${run.id}` }); } if (!['fail', 'blocked'].includes(nodeRun.status)) { return res.status(400).json({ error: `「重试」在节点 status='${nodeRun.status}' 下不可用,仅当节点 status ∈ {fail, blocked} 时可用`, op: 'retry', nodeStatus: nodeRun.status, allowed: ['fail', 'blocked'], }); } // ok:false / taken-by-other 是并发下的正常结果(§四.A.6 幂等:别人已经拿走就返回当前状态,不报错) res.json(engine.retryNode(run.id, nodeRun.id)); })); // ──────────────────────────────────────────────────────────────── // 只读查询类端点 // ──────────────────────────────────────────────────────────────── /** * GET /api/graph-runs/:id — 一次 run 的全量状态 * 返回 { run, nodes, edges, loopRecords, childRuns, budget } * nodes = 定义层(name / systemAgentId / nodeKind)与运行时(status / attempt / childRunId …)合并后的节点 * edges = 定义层原样(含 condition / failReason / edgeKind),画布靠 edgeKind='backward' 画虚线 * loopRecords = 回退账本,画布上 `N/3` 徽标的数据来源(§四.B.2) * budget = root run 的 budget_json 解析结果,页头预算条用(§四.B.4);非 root run 也回这个值 */ router.get('/:id', h((req, res) => { const state = getRunState(req.params.id); if (!state) return res.status(404).json({ error: `graph run 不存在: ${req.params.id}` }); const rootRun = state.run.root_run_id === state.run.id ? state.run : getRun(state.run.root_run_id); let budget = null; try { budget = rootRun && rootRun.budget_json ? JSON.parse(rootRun.budget_json) : null; } catch { budget = null; // 预算条坏掉不该让整个 run 查不出来 } // R-19 输入输出面板的三个数据源。没有它们,控制台的「输入输出」tab 整块降级 —— // 契约声明与产物索引都在库里,界面上却看不见,等于又一次「实现在、零调用方」。 // edges[].contracts = 边上声明的 kind(哪一层的边查哪一层的契约) // companyRoot = 库里只存相对路径,根由代码推导(R-19.3),前端靠它拼绝对路径 // taskFiles = 本 task 已登记的产物实例;契约 × 实例的 join 在前端做 const layer = state.run.graph_kind === 'org' ? 'org' : 'work'; const edges = (state.edges || []).map((e) => { const contracts = e.id ? edgeContracts(layer, e.id) : []; return contracts.length ? { ...e, contracts } : e; }); res.json({ ...state, edges, companyRoot: companyRootDir(state.run.company_id), taskFiles: findFiles({ companyId: state.run.company_id, taskId: state.run.task_id }), loopRecords: listLoopRecords(state.run.id), childRuns: listChildRuns(state.run.id), budget, }); })); /** * GET /api/graph-runs/:id/nodes/:nodeRunId/transcript — 这个节点那次调用的 LLM 原文 * * 引擎按设计不存逐字输出(见 agent-transcript.js 顶部),这里**惰性读** Claude CLI * 自己的 transcript 文件:不进 SSE、不进节点行、不落第二份库,只是多一个读取入口。 * * ⚠️ transcript 按 (公司, 角色) 分,**不按节点分**。同一角色跑过多个节点时只能按 * started_at / completed_at 截时间窗口 —— 近似而非精确。响应里原样带上 * `window` 与 `windowed`,让前端把这个前提显式写给用户,而不是假装是节点级归属。 * * 一次性拉取(前端不轮询它):整份会话是几百 KB 量级,跟着 run 状态轮询会很贵。 */ router.get('/:id/nodes/:nodeRunId/transcript', h((req, res) => { const run = getRun(req.params.id); if (!run) return res.status(404).json({ error: `graph run 不存在: ${req.params.id}` }); const node = getRunNode(req.params.nodeRunId); if (!node || node.run_id !== run.id) { return res.status(404).json({ error: `节点 ${req.params.nodeRunId} 不属于 run ${run.id}` }); } // 本体(角色名)绑在**定义层**节点上,graph_run_nodes 里没有这一列 —— // 运行时行只有 def_node_id,systemAgentId 是 getRunState 合并定义层之后才有的。 const merged = (getRunState(run.id) || { nodes: [] }).nodes.find((n) => n.id === node.id); const role = merged && merged.systemAgentId; if (!role) { // subgraph 节点自己不调 agent(它的活由子 run 里的节点做),没有可读的原文 return res.json({ ok: false, reason: 'no-agent', role: null, hint: '这个节点没有绑定本体(subgraph 节点自己不调 agent)——原文在子图各节点上。', }); } const transcript = require('../services/agent-transcript'); // completed_at 为空 = 还在跑:until 留空,这样正在写入的内容也能看到 const since = node.started_at || null; const until = node.completed_at || null; let out; try { out = transcript.readEntries({ companyId: run.company_id, role, since, until }); } catch (err) { return res.status(500).json({ error: `读 transcript 失败: ${err.message}` }); } res.json({ ...out, role, window: { since, until }, // 目录路径原样回出去:找不到文件时,用户能自己去看那个路径存不存在 // (比一句「暂无数据」有用得多 —— 后者分不清「没说过话」和「我们没找到」) attemptHint: `本节点 attempt #${(node.attempt || 0) + 1}`, }); })); /** * GET /api/graph-runs/:id/stream — SSE,实时推该 run 的状态转换事件 * 客户端可带 Last-Event-ID header(或 ?lastEventId=)断线重放,语义与 devloop 的 /stream 一致。 * * graph-run-sse.js 由 T5 并行落地;还没到位时回 503 而不是让整个 server 起不来。 */ router.get('/:id/stream', (req, res) => { const run = getRun(req.params.id); if (!run) return res.status(404).json({ error: `graph run 不存在: ${req.params.id}` }); let sse; try { sse = require('../services/graph-run-sse'); } catch (err) { return res.status(503).json({ error: `SSE 通道未就绪(graph-run-sse 未加载): ${err.message}` }); } res.setHeader('Content-Type', 'text/event-stream'); res.setHeader('Cache-Control', 'no-cache'); res.setHeader('Connection', 'keep-alive'); res.flushHeaders?.(); const lastEventId = req.headers['last-event-id'] || req.query.lastEventId || null; sse.subscribe(run.id, res, { replayFromOffset: lastEventId != null ? lastEventId : undefined }); }); /** * GET /api/company/:companyId/graph-runs — 某公司的 run 列表 * query: rootOnly=1 只列 root run(v1 一家公司同时只有一个 active root run,见 §一非目标) * 公司无 run 时返回空数组,不报错(与 GET /api/company/:cid/org-graph 的处理一致)。 */ /** * 「同一 company 只允许一个 active root run」的取值域,与 * work-graph-db 的 idx_graph_runs_one_active_root 逐字一致 * (`WHERE parent_run_id IS NULL AND run_status IN ('ready','running','paused')`)。 * **注意 blocked 不在里面** —— blocked 的 run 不占坑,可以再下一个任务。 */ const ACTIVE_ROOT_RUN_STATUS = new Set(['ready', 'running', 'paused']); companyRouter.get('/:companyId/graph-runs', h((req, res) => { const rootOnly = req.query.rootOnly === '1' || req.query.rootOnly === 'true'; const runs = listCompanyRuns(req.params.companyId, { rootOnly }); // canDispatch 由**服务端**算:前端此前自己复述了一份取值域并且多写了一个 blocked, // 于是一家公司只要失败过一次,控制台上就再也下不了第二个任务(后端其实放行)。 // 规则只留一处,前端照抄结果即可(P3B 第三次真跑发现)。 const activeRoot = runs.find((r) => !r.parent_run_id && ACTIVE_ROOT_RUN_STATUS.has(r.run_status)); res.json({ companyId: req.params.companyId, runs, canDispatch: !activeRoot, activeRootRunId: activeRoot ? activeRoot.id : null, }); })); module.exports = router; module.exports.companyRouter = companyRouter;