From 6ae1f7653a0eaa6fab76510e7d0e5a9cb15e0bca Mon Sep 17 00:00:00 2001 From: yibo Date: Sun, 30 Aug 2026 23:32:21 +0800 Subject: [PATCH] feat(graph-engine): 卡住恢复闭环——僵尸节点清扫 + 强制判失败端点（42 号文档 Phase 1） --- docs/todo/system-design/42-org-work-graph-reliability-followups.md | 11 +++++++++-- server/index.js | 16 ++++++++++++++++ server/routes/graph-run.js | 34 ++++++++++++++++++++++++++++++++++ server/services/graph-engine.js | 99 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ server/tests/unit/graph-engine-orphan-sweep.test.js | 160 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ server/tests/unit/graph-run-force-fail.test.js | 160 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 478 insertions(+), 2 deletions(-) create mode 100644 server/tests/unit/graph-engine-orphan-sweep.test.js create mode 100644 server/tests/unit/graph-run-force-fail.test.js diff --git a/docs/todo/system-design/42-org-work-graph-reliability-followups.md b/docs/todo/system-design/42-org-work-graph-reliability-followups.md index 5371051..42fe969 100644 --- a/docs/todo/system-design/42-org-work-graph-reliability-followups.md +++ b/docs/todo/system-design/42-org-work-graph-reliability-followups.md @@ -1,6 +1,6 @@ # 42. Org/Work Graph 可靠性收尾 —— 39 号文档真跑复盘的待办 -> **状态**：📝 待开工，未实现。 +> **状态**：Phase 1 ✅ 已实现（2026-08-30）；Phase 2-4 📝 待开工。 > **日期**：2026-08-30 > **背景**：[39-boss-agent-task-supervision.md](./39-boss-agent-task-supervision.md) 的 P1-P5 落地后，用它做了一次真实端到端测试（建公司 → 下 goal → `ceo` 监督 → 真实跑出一个可玩的贪吃蛇游戏，Playwright 5/5 验证通过）。**1469 个单测全绿，这次真跑却测出至少 3 个单测完全没覆盖到的真问题**——这份文档是那次复盘沉淀下来的待办，按开工顺序分四个 phase。 > **读者须知**：每个 phase 的现状描述都来自那次真实测试里实际撞到的场景，不是推测；具体文件/行号落地前按惯例重新核实。 @@ -20,7 +20,14 @@ Phase 4：timeline.ndjson 承接 planning/reflection 轨迹 —— 结构性改 --- -## 二、Phase 1：卡住恢复闭环 +## 二、Phase 1：卡住恢复闭环 —— ✅ 已实现（2026-08-30） + +**落地结果**（与下面的设计一一对应，实现时没有偏离）： +- 1a：`graph-engine.js` 新增 `sweepOrphanedRunningNodes(cutoffIso)`，复用 `routeFailure` 的 `infra_flake` 分支；`server/index.js` 在 `PROCESS_STARTED_AT`（模块顶层捕获的本次进程启动时刻）挂上开机一次性调用，排在其余「开机补一次全量扫描」钩子之前。测试：`server/tests/unit/graph-engine-orphan-sweep.test.js`（4 例：孤儿转 infra_flake 原地重跑、晚于 cutoff 的不受影响、连续判孤儿撞硬顶转 blocked、无孤儿时安静 no-op）。 +- 1b：`graph-engine.js` 新增 `forceFailNode(runId, nodeRunId, failReason='operator_intervention')`；`graph-run.js` 新增 `POST /:id/nodes/:nodeId/force-fail`，只接受 `status='running'`。测试：`server/tests/unit/graph-run-force-fail.test.js`（8 例：正常转 fail 并可被 retry 接上、自定义 failReason 透传、五种非 running 状态各返 400、节点不存在返 404）。 +- 全量测试 1483/1487，零回归（剩余 4 个是核实过的 pre-existing 失败）。 + +--- ### 1a. 服务器重启的僵尸节点检测 diff --git a/server/index.js b/server/index.js index 14f81c3..a0d26b8 100644 --- a/server/index.js +++ b/server/index.js @@ -18,6 +18,11 @@ const app = express(); const server = http.createServer(app); const pluginManager = new PluginManager(); +// 42 号文档 Phase 1a：本次进程的启动时刻，早于它的 graph_run_nodes.status='running' 一律是 +// 上一个进程实例留下的孤儿（这个新进程刚起，不可能自己已经有节点在跑）。模块顶层捕获一次， +// 不要在 sweepOrphanedRunningNodes 调用点现取——那样每次重试都会把截止线往后挪。 +const PROCESS_STARTED_AT = new Date().toISOString(); + // ─── 中间件 ─── app.use(cors(config.cors)); app.use(cookieParser()); @@ -160,6 +165,17 @@ async function start() { const taskSettle = require('./services/task-settle'); taskSettle.resumePendingGoalValidations() .catch(err => console.error('[Startup] 重推 goal 验收失败:', err.message)); + // 42 号文档 Phase 1a：上一个进程实例被换代码重启杀掉时，正在跑的 graph_run_nodes + // 会留在 status='running'（背后进程已死，行本身不会自己变）——ceo 的两条唤醒路径都 + // 只认 status='blocked'，对这个状态完全不可见。开机先清一遍，早于下面几个「补一次 + // 全量扫描」的钩子，让被它们观察的下游（task/goal 是不是卡住了）看到的是清好的状态。 + try { + const engine = require('./services/graph-engine'); + const { scanned, swept } = engine.sweepOrphanedRunningNodes(PROCESS_STARTED_AT); + console.log(`[Startup] 🧟 孤儿 graph 节点清扫：扫到 ${scanned} 条，处理 ${swept.length} 条`); + } catch (err) { + console.error('[Startup] 孤儿 graph 节点清扫失败:', err.message); + } // T-mt622nil：上面那条只补「收敛钩子已经跑过、卡在验收/汇报半路」的。更前面那一步—— // 钩子压根没被触发过（进程重启 / 异常吞掉 / 事件丢失）——task 早到终态但 goal 永远停在 // active，只能靠人工 OKR 巡检才发现（e2e-demo 公司 G-mt2r62ia-73af 卡了 48 小时）。 diff --git a/server/routes/graph-run.js b/server/routes/graph-run.js index 937f188..2eb2ae9 100644 --- a/server/routes/graph-run.js +++ b/server/routes/graph-run.js @@ -377,6 +377,40 @@ router.post('/:id/nodes/:nodeId/retry', h((req, res) => { })); /** + * POST /api/graph-runs/:id/nodes/:nodeId/force-fail — 「强制判失败」（42 号文档 Phase 1b） + * body: { failReason?: string }（默认 'operator_intervention'，刻意不是 'infra_flake'—— + * 那个值有特殊语义：原地重跑、不烧计数，人工判定不该被误当成一次性抖动） + * 可用条件：节点 status = 'running' + * + * `retry` 端点管不到 running 的节点（那是它自己的既有约定，不是这次要改的）——这个端点补的 + * 正是那个缺口：人类或 ceo 判断"这个节点已经不可能再自己完成了"时的手动出口。转 fail 之后 + * 交给引擎既有的失败路由（selectFailEdge/handleBackwardEdge），不发明新路径。 + */ +router.post('/:id/nodes/:nodeId/force-fail', 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 (nodeRun.status !== 'running') { + return res.status(400).json({ + error: `「强制判失败」在节点 status='${nodeRun.status}' 下不可用，仅当节点 status = 'running' 时可用（fail/blocked 请用 retry 端点）`, + op: 'force-fail', + nodeStatus: nodeRun.status, + allowed: ['running'], + }); + } + + const { failReason } = req.body || {}; + res.json(engine.forceFailNode(run.id, nodeRun.id, failReason || undefined)); +})); + +/** * POST /api/graph-runs/:id/budget — 「追加预算」（运维操作，不在 §四.C 五个操作之列） * body: { total_agent_calls?, total_tokens?, wallclock_ms?, resume?: boolean } * 可用条件：run_status = 'blocked' diff --git a/server/services/graph-engine.js b/server/services/graph-engine.js index 4e79cc8..a867b32 100644 --- a/server/services/graph-engine.js +++ b/server/services/graph-engine.js @@ -2905,6 +2905,53 @@ function stopWallclockWatchdog() { } /** + * 42 号文档 Phase 1a：开机孤儿节点清扫。 + * + * `graph_run_nodes.status='running'` 这一行不代表背后真的有个活的进程——它只是「上次问到的 + * 状态」。服务器重启（比如换代码生效）会把正在跑的 Claude CLI 子进程连带杀掉，但这一行留在 + * `running`，永远不会自己变成 `blocked`：`ceo` 的两条唤醒路径（事件驱动 + 30 分钟兜底扫描） + * 判据都只认 `status='blocked'`，对这个状态完全不可见（39 号文档 §七.1 真实撞出来的）。 + * + * 不发明新状态：这本质上跟 `infra_flake`（基础设施打断，跟设计/实现对不对无关）是同一类事， + * 服务器重启只是这个桶里更严重的一个成员——原地重跑的熔断线（`INFRA_FLAKE_HARD_CAP`）现成能用。 + * 复用 `routeFailure` 的 infra_flake 分支，不另起一套判定/熔断逻辑；调用形状照抄 `advanceOne` + * 里「先 `transitionNode` 落 fail_reason，再取新鲜的 `nodeRun`/`def` 传给 `routeFailure`」那一段。 + * + * @param {string} cutoffIso 判定孤儿的时间线——`started_at` 早于这个时刻的 running 节点判定为 + * 孤儿（本次进程启动前就已经在跑，不可能是本进程自己起的）。调用方传「本次进程启动时间」； + * 测试用任意时刻构造场景。 + * @returns {{scanned:number, swept:Array<{runId:string, nodeRunId:string, infraRerun:boolean}>}} + */ +function sweepOrphanedRunningNodes(cutoffIso) { + const rows = getDb() + .prepare(`SELECT id, run_id FROM graph_run_nodes WHERE status = 'running' AND started_at IS NOT NULL AND started_at < ?`) + .all(cutoffIso); + + const swept = []; + for (const row of rows) { + try { + const run = getRun(row.run_id); + if (!run) continue; // run 行本身没了（不该发生，防御性跳过） + const def = defIndex(run); + + const changed = sse.transitionNode(row.run_id, row.id, 'running', 'fail', { + failReason: INFRA_FLAKE_REASON, + failNodeRef: null, outputRef: null, agentRunId: null, childRunId: null, completedAt: now(), + }, { summary: `开机孤儿节点清扫：服务器重启前处于 running，背后进程已不存在，判 infra_flake 交给既有熔断/重跑逻辑` }); + if (!changed) continue; // 0 行 = 已经被别的路径结算过了，不重复处理 + + const nodeRun = getRunNode(row.id); + const out = { activated: [], backwardEdgeId: null, loop: null, contractDiff: null, crossLayer: null, reset: [], infraRerun: false }; + routeFailure(row.run_id, def, nodeRun, out, null); + swept.push({ runId: row.run_id, nodeRunId: row.id, infraRerun: out.infraRerun }); + } catch (err) { + console.error(`[graph-engine] sweepOrphanedRunningNodes node=${row.id} 异常: ${err.message}`); + } + } + return { scanned: rows.length, swept }; +} + +/** * 扣减（「只增不减」，§四.A.3）。始终写 root run 那一行。 * ponytail: 读-改-写不是原子的，当前单进程串行驱动（runAuto 一次只推一个节点）下够用； * 升级路径是把自增下沉成一条 `UPDATE ... SET budget_json = json_set(...)` 的 SQL。 @@ -3439,6 +3486,52 @@ function retryNode(runId, nodeRunId, { reopenRun = true, remediationNote = null }; } +/** + * 42 号文档 Phase 1b：「强制判失败以便恢复」。 + * + * `retryNode` 只接受 `fail`/`blocked` 的节点——但真实运维里会遇到一种它管不到的情形：节点 + * `status='running'`，人（或 `ceo`）判断它已经不可能再自己完成了（比如背后进程确实卡死， + * 不是 Phase 1a 那种"重启后一定是孤儿"，而是进程本身还在但没有任何进展）。在这条修复之前， + * 唯一的恢复手段是直接调 `graph-run-sse.transitionNode` 这个内部函数——对人类操作者或 `ceo` + * 都不现实，两者都只能走 HTTP。 + * + * 语义："我确认这个节点救不回来了，把它判 fail，交给既有的 fail→retry/backward-edge 路径 + * 处理"——不发明新的路由逻辑，`routeFailure` 该怎么选边还是怎么选边（默认原因不是 + * `infra_flake`，所以不会走原地重跑那条特判，是走 `selectFailEdge`/`handleBackwardEdge` + * 的常规路径）。 + * + * @param {string} runId + * @param {string} nodeRunId + * @param {string} [failReason='operator_intervention'] 默认值刻意不是 `infra_flake`——那个 + * 值有特殊语义（原地重跑、不烧计数），人工判定"这个真的救不回来了"不该被误当成一次性抖动。 + * @returns {{ok:boolean, reason?:string, nodeStatus?:string, runId?:string, nodeRunId?:string}} + */ +function forceFailNode(runId, nodeRunId, failReason = 'operator_intervention') { + const nodeRun = getRunNode(nodeRunId); + if (!nodeRun || nodeRun.run_id !== runId) throw new Error(`节点不属于该 run: ${nodeRunId}`); + if (nodeRun.status !== 'running') { + return { ok: false, reason: 'not-force-failable', nodeStatus: nodeRun.status }; + } + + const changed = sse.transitionNode(runId, nodeRunId, 'running', 'fail', { + failReason, + failNodeRef: null, outputRef: null, agentRunId: null, childRunId: null, completedAt: now(), + }, { summary: `人工「强制判失败」：running → fail（failReason=${failReason}），交给既有失败路由处理` }); + if (!changed) return { ok: false, reason: 'taken-by-other', nodeStatus: (getRunNode(nodeRunId) || {}).status }; + + const run = getRun(runId); + const def = defIndex(run); + const fresh = getRunNode(nodeRunId); + const out = { activated: [], backwardEdgeId: null, loop: null, contractDiff: null, crossLayer: null, reset: [], infraRerun: false }; + routeFailure(runId, def, fresh, out, null); + + return { + ok: true, runId, nodeRunId, + nodeStatus: (getRunNode(nodeRunId) || {}).status, + runStatus: (getRun(runId) || {}).run_status, + }; +} + // ──────────────────────────────────────────────────────────────── // §四.C 末段：重试后级联下游的判据 = contract_hash 变没变 // ──────────────────────────────────────────────────────────────── @@ -4775,6 +4868,9 @@ module.exports = { descendantRuns, landPause, + // 42 号文档 Phase 1b：running 节点的人工/ceo 强制判失败出口（retryNode 管不到 running） + forceFailNode, + // 任务级总预算（§四.A.3） checkBudget, spendBudget, @@ -4793,6 +4889,9 @@ module.exports = { startWallclockWatchdog, stopWallclockWatchdog, + // 42 号文档 Phase 1a：开机孤儿节点清扫（服务器重启杀掉 in-flight 进程，节点行留在 running） + sweepOrphanedRunningNodes, + // 重试级联判据（§四.C 末段） diffContracts, applyRetryCascade, diff --git a/server/tests/unit/graph-engine-orphan-sweep.test.js b/server/tests/unit/graph-engine-orphan-sweep.test.js new file mode 100644 index 0000000..c0688a2 --- /dev/null +++ b/server/tests/unit/graph-engine-orphan-sweep.test.js @@ -0,0 +1,160 @@ +/** + * Unit Tests — graph-engine 的开机孤儿节点清扫（42 号文档 Phase 1a） + * + * 场景：服务器进程被重启（比如换代码生效），正在跑某个 work graph 节点的真实 Claude CLI + * 子进程被连带杀掉，但 `graph_run_nodes.status` 永远停在 'running'——不会自己变成 'blocked'， + * ceo 的两条唤醒路径都只认 'blocked'，对这个状态完全不可见。 + * + * `sweepOrphanedRunningNodes(cutoffIso)` 扫一遍 `started_at` 早于 cutoff 的 running 节点， + * 判定为孤儿，走 infra_flake 既有的原地重跑/熔断逻辑（不发明新状态）。 + * + * 隔离手法与 graph-engine.test.js 一致：独立 GRAPH_DB_PATH + GRAPH_RUNS_DIR，真实 + * data/graph.sqlite / data/companies/ 全程不被碰到。 + */ +import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; +import fs from 'fs'; +import os from 'os'; +import path from 'path'; + +describe('graph-engine 开机孤儿节点清扫', () => { + let wdb; + let engine; + let runsDir; + const TEST_DB_PATH = '/tmp/test-graph-engine-orphan-sweep.sqlite'; + + beforeAll(() => { + for (const f of [TEST_DB_PATH, `${TEST_DB_PATH}-shm`, `${TEST_DB_PATH}-wal`]) { + try { fs.unlinkSync(f); } catch {} + } + process.env.GRAPH_DB_PATH = TEST_DB_PATH; + runsDir = fs.mkdtempSync(path.join(os.tmpdir(), 'graph-engine-orphan-sweep-')); + process.env.GRAPH_RUNS_DIR = runsDir; + for (const m of [ + '../../services/graph-db', + '../../services/graph-service', + '../../services/org-graph-service', + '../../services/work-graph-db', + '../../services/graph-run-sse', + '../../services/graph-engine', + ]) { + delete require.cache[require.resolve(m)]; + } + wdb = require('../../services/work-graph-db'); + engine = require('../../services/graph-engine'); + }); + + afterAll(() => { + for (const f of [TEST_DB_PATH, `${TEST_DB_PATH}-shm`, `${TEST_DB_PATH}-wal`]) { + try { fs.unlinkSync(f); } catch {} + } + delete process.env.GRAPH_DB_PATH; + try { fs.rmSync(runsDir, { recursive: true, force: true }); } catch {} + delete process.env.GRAPH_RUNS_DIR; + }); + + beforeEach(() => { + const db = wdb.getDb(); + db.pragma('foreign_keys = OFF'); + db.exec(` + DELETE FROM graph_run_loop_records; + DELETE FROM graph_run_nodes; + DELETE FROM graph_runs; + DELETE FROM work_graph_edges; + DELETE FROM work_graph_nodes; + DELETE FROM work_graphs; + DELETE FROM system_agents; + `); + db.pragma('foreign_keys = ON'); + engine.setAgentExecutor(null); + }); + + let seq = 0; + const nextCompany = () => `co-orphan-${++seq}`; + + function ensureAgents(ids) { + const db = wdb.getDb(); + const now = new Date().toISOString(); + const ins = db.prepare(` + INSERT OR IGNORE INTO system_agents + (id, name, scope, owner_company_id, legacy_path, status, created_at, updated_at) + VALUES (?, ?, 'public', NULL, ?, 'active', ?, ?) + `); + for (const id of new Set(ids)) ins.run(id, id, `system/agents/${id}`, now, now); + } + + function buildWorkRun({ companyId, nodes, edges }) { + ensureAgents(nodes); + const g = wdb.createWorkGraph({ companyId, nodes, edges }); + expect(g.errors || []).toEqual([]); + const run = wdb.createRun({ graphKind: 'work', graphId: g.id, companyId, executionMode: 'manual' }); + engine.markRunReady(run.id); + return { runId: run.id }; + } + + const nodeByName = (runId, name) => wdb.getRunState(runId).nodes.find((n) => n.name === name); + + /** 把某个节点直接摁成「孤儿」的静止态：status=running + 指定的 started_at，不走任何转换。 + * 这就是真实场景本身——进程被杀时节点行不会经过任何转换，就是停在这个样子。 */ + function markAsOrphan(runId, nodeName, startedAtIso) { + const node = nodeByName(runId, nodeName); + wdb.getDb() + .prepare(`UPDATE graph_run_nodes SET status = 'running', started_at = ?, updated_at = ? WHERE id = ?`) + .run(startedAtIso, startedAtIso, node.id); + return node.id; + } + + it('started_at 早于 cutoff 的 running 节点被判定为孤儿，走 infra_flake 原地重跑', () => { + const companyId = nextCompany(); + const { runId } = buildWorkRun({ companyId, nodes: ['w'], edges: [] }); + markAsOrphan(runId, 'w', '2020-01-01T00:00:00.000Z'); + + const result = engine.sweepOrphanedRunningNodes('2026-01-01T00:00:00.000Z'); + + expect(result.scanned).toBe(1); + expect(result.swept).toHaveLength(1); + expect(result.swept[0]).toMatchObject({ runId, infraRerun: true }); + // infra_flake 的原地重跑：转回 ready，不烧 attempt（§四.D.2.1 末段既有语义） + const fresh = nodeByName(runId, 'w'); + expect(fresh.status).toBe('ready'); + expect(fresh.attempt).toBe(0); + // 不写回退记录、不加 run 级 backward_count —— infra_flake 分支的既有保证，孤儿清扫复用 + // 的是同一条分支，这两条也该原样成立 + expect(wdb.listLoopRecords(runId)).toHaveLength(0); + expect(wdb.getRun(runId).backward_count).toBe(0); + }); + + it('started_at 晚于 cutoff 的 running 节点不受影响（真的还在跑，不是孤儿）', () => { + const companyId = nextCompany(); + const { runId } = buildWorkRun({ companyId, nodes: ['w'], edges: [] }); + markAsOrphan(runId, 'w', '2026-06-01T00:00:00.000Z'); + + const result = engine.sweepOrphanedRunningNodes('2026-01-01T00:00:00.000Z'); + + expect(result.scanned).toBe(0); + expect(result.swept).toHaveLength(0); + expect(nodeByName(runId, 'w').status).toBe('running'); + }); + + it('连续判定为孤儿超过 infra_flake 硬顶就转 blocked，不会无限原地重跑', () => { + const companyId = nextCompany(); + const { runId } = buildWorkRun({ companyId, nodes: ['w'], edges: [] }); + + // 连续三次「摁成孤儿再扫」，模拟每次重启后又立刻被杀的极端情形 + for (let i = 0; i < 3; i++) { + markAsOrphan(runId, 'w', '2020-01-01T00:00:00.000Z'); + engine.sweepOrphanedRunningNodes('2026-01-01T00:00:00.000Z'); + } + // 第四次撞硬顶（INFRA_FLAKE_HARD_CAP = 3），应改判 blocked 而不是继续 ready + markAsOrphan(runId, 'w', '2020-01-01T00:00:00.000Z'); + engine.sweepOrphanedRunningNodes('2026-01-01T00:00:00.000Z'); + + expect(nodeByName(runId, 'w').status).toBe('blocked'); + }); + + it('没有任何孤儿节点时是安静的 no-op', () => { + const companyId = nextCompany(); + buildWorkRun({ companyId, nodes: ['w'], edges: [] }); // 节点是 ready，不是 running + const result = engine.sweepOrphanedRunningNodes('2026-01-01T00:00:00.000Z'); + expect(result).toEqual({ scanned: 0, swept: [] }); + }); +}); diff --git a/server/tests/unit/graph-run-force-fail.test.js b/server/tests/unit/graph-run-force-fail.test.js new file mode 100644 index 0000000..b22e2c0 --- /dev/null +++ b/server/tests/unit/graph-run-force-fail.test.js @@ -0,0 +1,160 @@ +/** + * Unit Tests — POST /api/graph-runs/:id/nodes/:nodeId/force-fail（42 号文档 Phase 1b） + * + * `retry` 端点只接受 status ∈ {fail, blocked} 的节点，管不到 'running'——真实运维/ceo 遇到 + * "这个节点已经不可能再自己完成了"这种情形，此前唯一的恢复手段是直接调内部函数，绕过了 + * 任何 REST 接口。这条测试钉住新端点的可用条件与转移语义，不重复测 routeFailure 内部选边 + * 逻辑（那部分已经在 graph-engine.test.js / graph-engine-orphan-sweep.test.js 里覆盖过）。 + * + * 走真入口：把 router 挂到真 express 上 listen(0) 再 fetch，与 graph-run-route-io.test.js 同法。 + */ +import { describe, it, expect, beforeAll, afterAll } from 'vitest'; +import fs from 'fs'; +import os from 'os'; +import path from 'path'; +import express from 'express'; + +const TEST_DB_PATH = '/tmp/test-graph-run-force-fail.sqlite'; +const DB_FILES = [TEST_DB_PATH, `${TEST_DB_PATH}-shm`, `${TEST_DB_PATH}-wal`]; +const MODULES = [ + '../../services/graph-db', + '../../services/graph-service', + '../../services/org-graph-service', + '../../services/work-graph-db', + '../../services/company-paths', + '../../services/graph-run-sse', + '../../services/graph-engine', + '../../routes/graph-run', +]; + +describe('POST /api/graph-runs/:id/nodes/:nodeId/force-fail', () => { + let wdb; + let srv; + let base; + let runsDir; + let seq = 0; + // 每个用例一个独立 company：一家公司同时只有一条活跃 root run（run-slot 约束）， + // 同一个 company 起第二条 run 会撞 RunSlotBusyError。 + const nextCompany = () => `co-force-fail-${++seq}`; + + beforeAll(async () => { + for (const f of DB_FILES) { try { fs.unlinkSync(f); } catch {} } + process.env.GRAPH_DB_PATH = TEST_DB_PATH; + runsDir = fs.mkdtempSync(path.join(os.tmpdir(), 'force-fail-runs-')); + process.env.GRAPH_RUNS_DIR = runsDir; + for (const m of MODULES) delete require.cache[require.resolve(m)]; + wdb = require('../../services/work-graph-db'); + + const app = express(); + app.use(express.json()); + app.use('/api/graph-runs', require('../../routes/graph-run')); + await new Promise((resolve) => { + srv = app.listen(0, '127.0.0.1', () => { + base = `http://127.0.0.1:${srv.address().port}`; + resolve(); + }); + }); + }); + + afterAll(async () => { + if (srv) await new Promise((r) => srv.close(r)); + const { removeTestCompany } = require('../helpers/ensure-test-company'); + for (let i = 1; i <= seq; i++) removeTestCompany(`co-force-fail-${i}`); + for (const f of DB_FILES) { try { fs.unlinkSync(f); } catch {} } + delete process.env.GRAPH_DB_PATH; + try { fs.rmSync(runsDir, { recursive: true, force: true }); } catch {} + delete process.env.GRAPH_RUNS_DIR; + for (const m of MODULES) delete require.cache[require.resolve(m)]; + }); + + function ensureAgents(ids) { + const db = wdb.getDb(); + const now = new Date().toISOString(); + const ins = db.prepare(` + INSERT OR IGNORE INTO system_agents + (id, name, scope, owner_company_id, legacy_path, status, created_at, updated_at) + VALUES (?, ?, 'public', NULL, ?, 'active', ?, ?) + `); + for (const id of new Set(ids)) ins.run(id, id, `system/agents/${id}`, now, now); + } + + /** 建一条只有一个节点的 work run，直接把节点摁成指定状态（不关心怎么走到这个状态， + * 只关心端点在这个状态下的行为——执行语义已经在别的测试文件覆盖过）。 */ + function buildRunWithNodeStatus(status) { + const companyId = nextCompany(); + require('../helpers/ensure-test-company').ensureTestCompany(companyId); + ensureAgents(['w']); + const g = wdb.createWorkGraph({ companyId, nodes: ['w'], edges: [] }); + expect(g.errors || []).toEqual([]); + const run = wdb.createRun({ graphKind: 'work', graphId: g.id, companyId, executionMode: 'manual' }); + require('../../services/graph-engine').markRunReady(run.id); + const node = wdb.getRunState(run.id).nodes[0]; + const now = new Date().toISOString(); + wdb.getDb() + .prepare(`UPDATE graph_run_nodes SET status = ?, started_at = ?, updated_at = ? WHERE id = ?`) + .run(status, now, now, node.id); + return { runId: run.id, nodeRunId: node.id }; + } + + it('running 节点调用成功，转 fail 后交给既有失败路由（无边可走 → blocked，可被 retry 接上）', async () => { + const { runId, nodeRunId } = buildRunWithNodeStatus('running'); + const r = await fetch(`${base}/api/graph-runs/${runId}/nodes/${nodeRunId}/force-fail`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({}), + }); + expect(r.status).toBe(200); + const body = await r.json(); + expect(body.ok).toBe(true); + // 单节点图没有 fail 边可走 → blockNode，落在人工可 retry 的终点 + expect(body.nodeStatus).toBe('blocked'); + + // 验证确实接得上 retry 端点（这才是「可被 retry 端点接上」这条验收的真实含义） + const retryRes = await fetch(`${base}/api/graph-runs/${runId}/nodes/${nodeRunId}/retry`, { method: 'POST' }); + expect(retryRes.status).toBe(200); + const retryBody = await retryRes.json(); + expect(retryBody.ok).toBe(true); + expect(retryBody.nodeStatus).toBe('ready'); + }); + + it('自定义 failReason 会原样落进 fail_reason（走常规选边，不是 infra_flake 特判）', async () => { + const { runId, nodeRunId } = buildRunWithNodeStatus('running'); + const r = await fetch(`${base}/api/graph-runs/${runId}/nodes/${nodeRunId}/force-fail`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ failReason: 'stuck_no_progress' }), + }); + expect(r.status).toBe(200); + // 单节点图没有 fail 边，无论什么 fail_reason 都落 blocked——这里只确认端点把自定义原因 + // 传下去了没炸，选边逻辑本身已经在别处测过 + const body = await r.json(); + expect(body.ok).toBe(true); + expect(body.nodeStatus).toBe('blocked'); + }); + + for (const status of ['fail', 'blocked', 'pending', 'ready', 'success']) { + it(`节点 status='${status}' 时调用返回 400（不是 running 就不该走这个端点）`, async () => { + const { runId, nodeRunId } = buildRunWithNodeStatus(status); + const r = await fetch(`${base}/api/graph-runs/${runId}/nodes/${nodeRunId}/force-fail`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({}), + }); + expect(r.status).toBe(400); + const body = await r.json(); + expect(body.op).toBe('force-fail'); + expect(body.nodeStatus).toBe(status); + expect(body.allowed).toEqual(['running']); + }); + } + + it('节点不存在返回 404', async () => { + const { runId } = buildRunWithNodeStatus('running'); + const r = await fetch(`${base}/api/graph-runs/${runId}/nodes/nope/force-fail`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({}), + }); + expect(r.status).toBe(404); + }); +}); -- libgit2 0.21.4