宜博 / lemonboss

Commit 6ae1f7653a0eaa6fab76510e7d0e5a9cb15e0bca

Authored by yibo
1 parent 5f2560da

feat(graph-engine): 卡住恢复闭环——僵尸节点清扫 + 强制判失败端点（42 号文档 Phase 1） 

服务器重启会杀死正在跑的 graph 节点背后的真实进程，但 graph_run_nodes.
status 永远停在 running（ceo 的两条唤醒路径都只认 blocked，对这个状态
完全不可见）——39 号文档真跑复盘撞出来的坑，这次补上。

1a：graph-engine.js 新增 sweepOrphanedRunningNodes(cutoffIso)，复用
routeFailure 的 infra_flake 分支（不发明新状态），server/index.js 在
本次进程启动时刻挂一次开机扫描，排在其余「补一次全量扫描」钩子之前。

1b：新增 forceFailNode() + POST /:id/nodes/:nodeId/force-fail，补上
retry 端点管不到 running 节点这个缺口——人工/ceo 判定"这个真的救不
回来了"时的手动出口，转 fail 后交给既有失败路由处理，不发明新路径。

新增测试 12 例（graph-engine-orphan-sweep.test.js 4 例、
graph-run-force-fail.test.js 8 例），全量测试 1483/1487，零回归。

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BwG4b7Nq1NLR9eeE3M9Us9
Inline Side-by-side
Showing 6 changed files with 478 additions and 2 deletions
docs/todo/system-design/42-org-work-graph-reliability-followups.md
  View file @6ae1f76
1 1 # 42. Org/Work Graph å¯é æ€§æ”¶å°¾ —— 39 å·æ–‡æ¡£çœŸè·‘å¤ç›˜çš„待办
2 2  
3   -> **状æ€**：📠待开工，未实现。
  3 +> **状æ€**：Phase 1 ✅ 已实现（2026-08-30）；Phase 2-4 📠待开工。
4 4 > **日期**：2026-08-30
5 5 > **背景**：[39-boss-agent-task-supervision.md](./39-boss-agent-task-supervision.md) çš„ P1-P5 è½åœ°åŽï¼Œç”¨å®ƒåšäº†ä¸€æ¬¡çœŸå®žç«¯åˆ°ç«¯æµ‹è¯•ï¼ˆå»ºå…¬å¸ → 下 goal → `ceo` ç›‘ç£ â†’ 真实跑出一个å¯çŽ©çš„è´ªåƒè›‡æ¸¸æˆï¼ŒPlaywright 5/5 验è¯é€šè¿‡ï¼‰ã€‚**1469 个å•æµ‹å…¨ç»¿ï¼Œè¿™æ¬¡çœŸè·‘å´æµ‹å‡ºè‡³å°‘ 3 个å•æµ‹å®Œå…¨æ²¡è¦†ç›–到的真问题**——这份文档是那次å¤ç›˜æ²‰æ·€ä¸‹æ¥çš„待办，按开工顺åºåˆ†å››ä¸ª phase。
6 6 > **读者须知**：æ¯ä¸ª phase 的现状æ述都æ¥è‡ªé‚£æ¬¡çœŸå®žæµ‹è¯•é‡Œå®žé™…撞到的场景，ä¸æ˜¯æŽ¨æµ‹ï¼›å…·ä½“文件/è¡Œå·è½åœ°å‰æŒ‰æƒ¯ä¾‹é‡æ–°æ ¸å®žã€‚
... ... @@ -20,7 +20,14 @@ Phase 4：timeline.ndjson 承接 planning/reflection 轨迹 —— 结构性改å
20 20  
21 21 ---
22 22  
23   -## 二ã€Phase 1：å¡ä½æ¢å¤é—­çŽ¯
  23 +## 二ã€Phase 1：å¡ä½æ¢å¤é—­çŽ¯ —— ✅ 已实现（2026-08-30）
  24 +
  25 +**è½åœ°ç»“æžœ**（与下é¢çš„设计一一对应，实现时没有å离）：
  26 +- 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）。
  27 +- 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）。
  28 +- å…¨é‡æµ‹è¯• 1483/1487，零回归（剩余 4 个是核实过的 pre-existing 失败）。
  29 +
  30 +---
24 31  
25 32 ### 1a. æœåŠ¡å™¨é‡å¯çš„僵尸节点检测
26 33  
... ...
server/index.js
  View file @6ae1f76
... ... @@ -18,6 +18,11 @@ const app = express();
18 18 const server = http.createServer(app);
19 19 const pluginManager = new PluginManager();
20 20  
  21 +// 42 号文档 Phase 1a：本次进程的启动时刻，早于它的 graph_run_nodes.status='running' 一律是
  22 +// 上一个进程实例留下的孤儿（这个新进程刚起，不可能自己已经有节点在跑）。模块顶层捕获一次，
  23 +// 不要在 sweepOrphanedRunningNodes 调用点现取——那样每次重试都会把截止线往后挪。
  24 +const PROCESS_STARTED_AT = new Date().toISOString();
  25 +
21 26 // ─── 中间件 ───
22 27 app.use(cors(config.cors));
23 28 app.use(cookieParser());
... ... @@ -160,6 +165,17 @@ async function start() {
160 165 const taskSettle = require('./services/task-settle');
161 166 taskSettle.resumePendingGoalValidations()
162 167 .catch(err => console.error('[Startup] 重推 goal 验收失败:', err.message));
  168 + // 42 号文档 Phase 1a：上一个进程实例被换代码重启杀掉时，正在跑的 graph_run_nodes
  169 + // 会留在 status='running'（背后进程已死，行本身不会自己变）——ceo 的两条唤醒路径都
  170 + // 只认 status='blocked'，对这个状态完全不可见。开机先清一遍，早于下面几个「补一次
  171 + // 全量扫描」的钩子，让被它们观察的下游（task/goal 是不是卡住了）看到的是清好的状态。
  172 + try {
  173 + const engine = require('./services/graph-engine');
  174 + const { scanned, swept } = engine.sweepOrphanedRunningNodes(PROCESS_STARTED_AT);
  175 + console.log(`[Startup] 🧟 孤儿 graph 节点清扫：扫到 ${scanned} 条，处理 ${swept.length} 条`);
  176 + } catch (err) {
  177 + console.error('[Startup] 孤儿 graph 节点清扫失败:', err.message);
  178 + }
163 179 // T-mt622nil：上面那条只补「收敛钩子已经跑过、卡在验收/汇报半路」的。更前面那一步——
164 180 // 钩子压根没被触发过（进程重启 / 异常吞掉 / 事件丢失）——task 早到终态但 goal 永远停在
165 181 // active，只能靠人工 OKR 巡检才发现（e2e-demo 公司 G-mt2r62ia-73af 卡了 48 小时）。
... ...
server/routes/graph-run.js
  View file @6ae1f76
... ... @@ -377,6 +377,40 @@ router.post(&#39;/:id/nodes/:nodeId/retry&#39;, h((req, res) =&gt; {
377 377 }));
378 378  
379 379 /**
  380 + * POST /api/graph-runs/:id/nodes/:nodeId/force-fail — 「强制判失败」（42 号文档 Phase 1b）
  381 + * body: { failReason?: string }（默认 'operator_intervention'，刻意不是 'infra_flake'——
  382 + * 那个值有特殊语义：原地重跑、不烧计数，人工判定不该被误当成一次性抖动）
  383 + * 可用条件：节点 status = 'running'
  384 + *
  385 + * `retry` 端点管不到 running 的节点（那是它自己的既有约定，不是这次要改的）——这个端点补的
  386 + * 正是那个缺口：人类或 ceo 判断"这个节点已经不可能再自己完成了"时的手动出口。转 fail 之后
  387 + * 交给引擎既有的失败路由（selectFailEdge/handleBackwardEdge），不发明新路径。
  388 + */
  389 +router.post('/:id/nodes/:nodeId/force-fail', h((req, res) => {
  390 + const run = loadRun(req, res);
  391 + if (!run) return;
  392 +
  393 + const { nodeId } = req.params;
  394 + let nodeRun = getRunNode(nodeId);
  395 + if (!nodeRun || nodeRun.run_id !== run.id) nodeRun = getRunNodeByDefId(run.id, nodeId);
  396 + if (!nodeRun) {
  397 + return res.status(404).json({ error: `节点 ${nodeId} 不属于 run ${run.id}` });
  398 + }
  399 +
  400 + if (nodeRun.status !== 'running') {
  401 + return res.status(400).json({
  402 + error: `「强制判失败」在节点 status='${nodeRun.status}' 下不可用，仅当节点 status = 'running' 时可用（fail/blocked 请用 retry 端点）`,
  403 + op: 'force-fail',
  404 + nodeStatus: nodeRun.status,
  405 + allowed: ['running'],
  406 + });
  407 + }
  408 +
  409 + const { failReason } = req.body || {};
  410 + res.json(engine.forceFailNode(run.id, nodeRun.id, failReason || undefined));
  411 +}));
  412 +
  413 +/**
380 414 * POST /api/graph-runs/:id/budget — 「追加预算」（运维操作，不在 §四.C 五个操作之列）
381 415 * body: { total_agent_calls?, total_tokens?, wallclock_ms?, resume?: boolean }
382 416 * 可用条件：run_status = 'blocked'
... ...
server/services/graph-engine.js
  View file @6ae1f76
... ... @@ -2905,6 +2905,53 @@ function stopWallclockWatchdog() {
2905 2905 }
2906 2906  
2907 2907 /**
  2908 + * 42 号文档 Phase 1a：开机孤儿节点清扫。
  2909 + *
  2910 + * `graph_run_nodes.status='running'` 这一行不代表背后真的有个活的进程——它只是「上次问到的
  2911 + * 状态」。服务器重启（比如换代码生效）会把正在跑的 Claude CLI 子进程连带杀掉，但这一行留在
  2912 + * `running`，永远不会自己变成 `blocked`：`ceo` 的两条唤醒路径（事件驱动 + 30 分钟兜底扫描）
  2913 + * 判据都只认 `status='blocked'`，对这个状态完全不可见（39 号文档 §七.1 真实撞出来的）。
  2914 + *
  2915 + * 不发明新状态：这本质上跟 `infra_flake`（基础设施打断，跟设计/实现对不对无关）是同一类事，
  2916 + * 服务器重启只是这个桶里更严重的一个成员——原地重跑的熔断线（`INFRA_FLAKE_HARD_CAP`）现成能用。
  2917 + * 复用 `routeFailure` 的 infra_flake 分支，不另起一套判定/熔断逻辑；调用形状照抄 `advanceOne`
  2918 + * 里「先 `transitionNode` 落 fail_reason，再取新鲜的 `nodeRun`/`def` 传给 `routeFailure`」那一段。
  2919 + *
  2920 + * @param {string} cutoffIso 判定孤儿的时间线——`started_at` 早于这个时刻的 running 节点判定为
  2921 + * 孤儿（本次进程启动前就已经在跑，不可能是本进程自己起的）。调用方传「本次进程启动时间」；
  2922 + * 测试用任意时刻构造场景。
  2923 + * @returns {{scanned:number, swept:Array<{runId:string, nodeRunId:string, infraRerun:boolean}>}}
  2924 + */
  2925 +function sweepOrphanedRunningNodes(cutoffIso) {
  2926 + const rows = getDb()
  2927 + .prepare(`SELECT id, run_id FROM graph_run_nodes WHERE status = 'running' AND started_at IS NOT NULL AND started_at < ?`)
  2928 + .all(cutoffIso);
  2929 +
  2930 + const swept = [];
  2931 + for (const row of rows) {
  2932 + try {
  2933 + const run = getRun(row.run_id);
  2934 + if (!run) continue; // run 行本身没了（不该发生，防御性跳过）
  2935 + const def = defIndex(run);
  2936 +
  2937 + const changed = sse.transitionNode(row.run_id, row.id, 'running', 'fail', {
  2938 + failReason: INFRA_FLAKE_REASON,
  2939 + failNodeRef: null, outputRef: null, agentRunId: null, childRunId: null, completedAt: now(),
  2940 + }, { summary: `开机孤儿节点清扫：服务器重启前处于 running，背后进程已不存在，判 infra_flake 交给既有熔断/重跑逻辑` });
  2941 + if (!changed) continue; // 0 行 = 已经被别的路径结算过了，不重复处理
  2942 +
  2943 + const nodeRun = getRunNode(row.id);
  2944 + const out = { activated: [], backwardEdgeId: null, loop: null, contractDiff: null, crossLayer: null, reset: [], infraRerun: false };
  2945 + routeFailure(row.run_id, def, nodeRun, out, null);
  2946 + swept.push({ runId: row.run_id, nodeRunId: row.id, infraRerun: out.infraRerun });
  2947 + } catch (err) {
  2948 + console.error(`[graph-engine] sweepOrphanedRunningNodes node=${row.id} 异常: ${err.message}`);
  2949 + }
  2950 + }
  2951 + return { scanned: rows.length, swept };
  2952 +}
  2953 +
  2954 +/**
2908 2955 * 扣减（「只增不减」，§四.A.3）。始终写 root run 那一行。
2909 2956 * ponytail: 读-改-写不是原子的，当前单进程串行驱动（runAuto 一次只推一个节点）下够用；
2910 2957 * 升级路径是把自增下沉成一条 `UPDATE ... SET budget_json = json_set(...)` 的 SQL。
... ... @@ -3439,6 +3486,52 @@ function retryNode(runId, nodeRunId, { reopenRun = true, remediationNote = null
3439 3486 };
3440 3487 }
3441 3488  
  3489 +/**
  3490 + * 42 号文档 Phase 1b：「强制判失败以便恢复」。
  3491 + *
  3492 + * `retryNode` 只接受 `fail`/`blocked` 的节点——但真实运维里会遇到一种它管不到的情形：节点
  3493 + * `status='running'`，人（或 `ceo`）判断它已经不可能再自己完成了（比如背后进程确实卡死，
  3494 + * 不是 Phase 1a 那种"重启后一定是孤儿"，而是进程本身还在但没有任何进展）。在这条修复之前，
  3495 + * 唯一的恢复手段是直接调 `graph-run-sse.transitionNode` 这个内部函数——对人类操作者或 `ceo`
  3496 + * 都不现实，两者都只能走 HTTP。
  3497 + *
  3498 + * 语义："我确认这个节点救不回来了，把它判 fail，交给既有的 fail→retry/backward-edge 路径
  3499 + * 处理"——不发明新的路由逻辑，`routeFailure` 该怎么选边还是怎么选边（默认原因不是
  3500 + * `infra_flake`，所以不会走原地重跑那条特判，是走 `selectFailEdge`/`handleBackwardEdge`
  3501 + * 的常规路径）。
  3502 + *
  3503 + * @param {string} runId
  3504 + * @param {string} nodeRunId
  3505 + * @param {string} [failReason='operator_intervention'] 默认值刻意不是 `infra_flake`——那个
  3506 + * 值有特殊语义（原地重跑、不烧计数），人工判定"这个真的救不回来了"不该被误当成一次性抖动。
  3507 + * @returns {{ok:boolean, reason?:string, nodeStatus?:string, runId?:string, nodeRunId?:string}}
  3508 + */
  3509 +function forceFailNode(runId, nodeRunId, failReason = 'operator_intervention') {
  3510 + const nodeRun = getRunNode(nodeRunId);
  3511 + if (!nodeRun || nodeRun.run_id !== runId) throw new Error(`节点不属于该 run: ${nodeRunId}`);
  3512 + if (nodeRun.status !== 'running') {
  3513 + return { ok: false, reason: 'not-force-failable', nodeStatus: nodeRun.status };
  3514 + }
  3515 +
  3516 + const changed = sse.transitionNode(runId, nodeRunId, 'running', 'fail', {
  3517 + failReason,
  3518 + failNodeRef: null, outputRef: null, agentRunId: null, childRunId: null, completedAt: now(),
  3519 + }, { summary: `人工「强制判失败」：running → fail（failReason=${failReason}），交给既有失败路由处理` });
  3520 + if (!changed) return { ok: false, reason: 'taken-by-other', nodeStatus: (getRunNode(nodeRunId) || {}).status };
  3521 +
  3522 + const run = getRun(runId);
  3523 + const def = defIndex(run);
  3524 + const fresh = getRunNode(nodeRunId);
  3525 + const out = { activated: [], backwardEdgeId: null, loop: null, contractDiff: null, crossLayer: null, reset: [], infraRerun: false };
  3526 + routeFailure(runId, def, fresh, out, null);
  3527 +
  3528 + return {
  3529 + ok: true, runId, nodeRunId,
  3530 + nodeStatus: (getRunNode(nodeRunId) || {}).status,
  3531 + runStatus: (getRun(runId) || {}).run_status,
  3532 + };
  3533 +}
  3534 +
3442 3535 // ────────────────────────────────────────────────────────────────
3443 3536 // §四.C 末段：重试后级联下游的判据 = contract_hash 变没变
3444 3537 // ────────────────────────────────────────────────────────────────
... ... @@ -4775,6 +4868,9 @@ module.exports = {
4775 4868 descendantRuns,
4776 4869 landPause,
4777 4870  
  4871 + // 42 号文档 Phase 1b：running 节点的人工/ceo 强制判失败出口（retryNode 管不到 running）
  4872 + forceFailNode,
  4873 +
4778 4874 // 任务级总预算（§四.A.3）
4779 4875 checkBudget,
4780 4876 spendBudget,
... ... @@ -4793,6 +4889,9 @@ module.exports = {
4793 4889 startWallclockWatchdog,
4794 4890 stopWallclockWatchdog,
4795 4891  
  4892 + // 42 号文档 Phase 1a：开机孤儿节点清扫（服务器重启杀掉 in-flight 进程，节点行留在 running）
  4893 + sweepOrphanedRunningNodes,
  4894 +
4796 4895 // 重试级联判据（§四.C 末段）
4797 4896 diffContracts,
4798 4897 applyRetryCascade,
... ...
server/tests/unit/graph-engine-orphan-sweep.test.js 0 → 100644
  View file @6ae1f76
  1 +/**
  2 + * Unit Tests — graph-engine 的开机孤儿节点清扫（42 号文档 Phase 1a）
  3 + *
  4 + * 场景：服务器进程被重启（比如换代码生效），正在跑某个 work graph 节点的真实 Claude CLI
  5 + * 子进程被连带杀掉，但 `graph_run_nodes.status` 永远停在 'running'——不会自己变成 'blocked'，
  6 + * ceo 的两条唤醒路径都只认 'blocked'，对这个状态完全不可见。
  7 + *
  8 + * `sweepOrphanedRunningNodes(cutoffIso)` 扫一遍 `started_at` 早于 cutoff 的 running 节点，
  9 + * 判定为孤儿，走 infra_flake 既有的原地重跑/熔断逻辑（不发明新状态）。
  10 + *
  11 + * 隔离手法与 graph-engine.test.js 一致：独立 GRAPH_DB_PATH + GRAPH_RUNS_DIR，真实
  12 + * data/graph.sqlite / data/companies/ 全程不被碰到。
  13 + */
  14 +import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
  15 +import fs from 'fs';
  16 +import os from 'os';
  17 +import path from 'path';
  18 +
  19 +describe('graph-engine 开机孤儿节点清扫', () => {
  20 + let wdb;
  21 + let engine;
  22 + let runsDir;
  23 + const TEST_DB_PATH = '/tmp/test-graph-engine-orphan-sweep.sqlite';
  24 +
  25 + beforeAll(() => {
  26 + for (const f of [TEST_DB_PATH, `${TEST_DB_PATH}-shm`, `${TEST_DB_PATH}-wal`]) {
  27 + try { fs.unlinkSync(f); } catch {}
  28 + }
  29 + process.env.GRAPH_DB_PATH = TEST_DB_PATH;
  30 + runsDir = fs.mkdtempSync(path.join(os.tmpdir(), 'graph-engine-orphan-sweep-'));
  31 + process.env.GRAPH_RUNS_DIR = runsDir;
  32 + for (const m of [
  33 + '../../services/graph-db',
  34 + '../../services/graph-service',
  35 + '../../services/org-graph-service',
  36 + '../../services/work-graph-db',
  37 + '../../services/graph-run-sse',
  38 + '../../services/graph-engine',
  39 + ]) {
  40 + delete require.cache[require.resolve(m)];
  41 + }
  42 + wdb = require('../../services/work-graph-db');
  43 + engine = require('../../services/graph-engine');
  44 + });
  45 +
  46 + afterAll(() => {
  47 + for (const f of [TEST_DB_PATH, `${TEST_DB_PATH}-shm`, `${TEST_DB_PATH}-wal`]) {
  48 + try { fs.unlinkSync(f); } catch {}
  49 + }
  50 + delete process.env.GRAPH_DB_PATH;
  51 + try { fs.rmSync(runsDir, { recursive: true, force: true }); } catch {}
  52 + delete process.env.GRAPH_RUNS_DIR;
  53 + });
  54 +
  55 + beforeEach(() => {
  56 + const db = wdb.getDb();
  57 + db.pragma('foreign_keys = OFF');
  58 + db.exec(`
  59 + DELETE FROM graph_run_loop_records;
  60 + DELETE FROM graph_run_nodes;
  61 + DELETE FROM graph_runs;
  62 + DELETE FROM work_graph_edges;
  63 + DELETE FROM work_graph_nodes;
  64 + DELETE FROM work_graphs;
  65 + DELETE FROM system_agents;
  66 + `);
  67 + db.pragma('foreign_keys = ON');
  68 + engine.setAgentExecutor(null);
  69 + });
  70 +
  71 + let seq = 0;
  72 + const nextCompany = () => `co-orphan-${++seq}`;
  73 +
  74 + function ensureAgents(ids) {
  75 + const db = wdb.getDb();
  76 + const now = new Date().toISOString();
  77 + const ins = db.prepare(`
  78 + INSERT OR IGNORE INTO system_agents
  79 + (id, name, scope, owner_company_id, legacy_path, status, created_at, updated_at)
  80 + VALUES (?, ?, 'public', NULL, ?, 'active', ?, ?)
  81 + `);
  82 + for (const id of new Set(ids)) ins.run(id, id, `system/agents/${id}`, now, now);
  83 + }
  84 +
  85 + function buildWorkRun({ companyId, nodes, edges }) {
  86 + ensureAgents(nodes);
  87 + const g = wdb.createWorkGraph({ companyId, nodes, edges });
  88 + expect(g.errors || []).toEqual([]);
  89 + const run = wdb.createRun({ graphKind: 'work', graphId: g.id, companyId, executionMode: 'manual' });
  90 + engine.markRunReady(run.id);
  91 + return { runId: run.id };
  92 + }
  93 +
  94 + const nodeByName = (runId, name) => wdb.getRunState(runId).nodes.find((n) => n.name === name);
  95 +
  96 + /** 把某个节点直接摁成「孤儿」的静止态：status=running + 指定的 started_at，不走任何转换。
  97 + * 这就是真实场景本身——进程被杀时节点行不会经过任何转换，就是停在这个样子。 */
  98 + function markAsOrphan(runId, nodeName, startedAtIso) {
  99 + const node = nodeByName(runId, nodeName);
  100 + wdb.getDb()
  101 + .prepare(`UPDATE graph_run_nodes SET status = 'running', started_at = ?, updated_at = ? WHERE id = ?`)
  102 + .run(startedAtIso, startedAtIso, node.id);
  103 + return node.id;
  104 + }
  105 +
  106 + it('started_at 早于 cutoff 的 running 节点被判定为孤儿，走 infra_flake 原地重跑', () => {
  107 + const companyId = nextCompany();
  108 + const { runId } = buildWorkRun({ companyId, nodes: ['w'], edges: [] });
  109 + markAsOrphan(runId, 'w', '2020-01-01T00:00:00.000Z');
  110 +
  111 + const result = engine.sweepOrphanedRunningNodes('2026-01-01T00:00:00.000Z');
  112 +
  113 + expect(result.scanned).toBe(1);
  114 + expect(result.swept).toHaveLength(1);
  115 + expect(result.swept[0]).toMatchObject({ runId, infraRerun: true });
  116 + // infra_flake 的原地重跑：转回 ready，不烧 attempt（§四.D.2.1 末段既有语义）
  117 + const fresh = nodeByName(runId, 'w');
  118 + expect(fresh.status).toBe('ready');
  119 + expect(fresh.attempt).toBe(0);
  120 + // 不写回退记录、不加 run 级 backward_count —— infra_flake 分支的既有保证，孤儿清扫复用
  121 + // 的是同一条分支，这两条也该原样成立
  122 + expect(wdb.listLoopRecords(runId)).toHaveLength(0);
  123 + expect(wdb.getRun(runId).backward_count).toBe(0);
  124 + });
  125 +
  126 + it('started_at 晚于 cutoff 的 running 节点不受影响（真的还在跑，不是孤儿）', () => {
  127 + const companyId = nextCompany();
  128 + const { runId } = buildWorkRun({ companyId, nodes: ['w'], edges: [] });
  129 + markAsOrphan(runId, 'w', '2026-06-01T00:00:00.000Z');
  130 +
  131 + const result = engine.sweepOrphanedRunningNodes('2026-01-01T00:00:00.000Z');
  132 +
  133 + expect(result.scanned).toBe(0);
  134 + expect(result.swept).toHaveLength(0);
  135 + expect(nodeByName(runId, 'w').status).toBe('running');
  136 + });
  137 +
  138 + it('连续判定为孤儿超过 infra_flake 硬顶就转 blocked，不会无限原地重跑', () => {
  139 + const companyId = nextCompany();
  140 + const { runId } = buildWorkRun({ companyId, nodes: ['w'], edges: [] });
  141 +
  142 + // 连续三次「摁成孤儿再扫」，模拟每次重启后又立刻被杀的极端情形
  143 + for (let i = 0; i < 3; i++) {
  144 + markAsOrphan(runId, 'w', '2020-01-01T00:00:00.000Z');
  145 + engine.sweepOrphanedRunningNodes('2026-01-01T00:00:00.000Z');
  146 + }
  147 + // 第四次撞硬顶（INFRA_FLAKE_HARD_CAP = 3），应改判 blocked 而不是继续 ready
  148 + markAsOrphan(runId, 'w', '2020-01-01T00:00:00.000Z');
  149 + engine.sweepOrphanedRunningNodes('2026-01-01T00:00:00.000Z');
  150 +
  151 + expect(nodeByName(runId, 'w').status).toBe('blocked');
  152 + });
  153 +
  154 + it('没有任何孤儿节点时是安静的 no-op', () => {
  155 + const companyId = nextCompany();
  156 + buildWorkRun({ companyId, nodes: ['w'], edges: [] }); // 节点是 ready，不是 running
  157 + const result = engine.sweepOrphanedRunningNodes('2026-01-01T00:00:00.000Z');
  158 + expect(result).toEqual({ scanned: 0, swept: [] });
  159 + });
  160 +});
... ...
server/tests/unit/graph-run-force-fail.test.js 0 → 100644
  View file @6ae1f76
  1 +/**
  2 + * Unit Tests — POST /api/graph-runs/:id/nodes/:nodeId/force-fail（42 号文档 Phase 1b）
  3 + *
  4 + * `retry` 端点只接受 status ∈ {fail, blocked} 的节点，管不到 'running'——真实运维/ceo 遇到
  5 + * "这个节点已经不可能再自己完成了"这种情形，此前唯一的恢复手段是直接调内部函数，绕过了
  6 + * 任何 REST 接口。这条测试钉住新端点的可用条件与转移语义，不重复测 routeFailure 内部选边
  7 + * 逻辑（那部分已经在 graph-engine.test.js / graph-engine-orphan-sweep.test.js 里覆盖过）。
  8 + *
  9 + * 走真入口：把 router 挂到真 express 上 listen(0) 再 fetch，与 graph-run-route-io.test.js 同法。
  10 + */
  11 +import { describe, it, expect, beforeAll, afterAll } from 'vitest';
  12 +import fs from 'fs';
  13 +import os from 'os';
  14 +import path from 'path';
  15 +import express from 'express';
  16 +
  17 +const TEST_DB_PATH = '/tmp/test-graph-run-force-fail.sqlite';
  18 +const DB_FILES = [TEST_DB_PATH, `${TEST_DB_PATH}-shm`, `${TEST_DB_PATH}-wal`];
  19 +const MODULES = [
  20 + '../../services/graph-db',
  21 + '../../services/graph-service',
  22 + '../../services/org-graph-service',
  23 + '../../services/work-graph-db',
  24 + '../../services/company-paths',
  25 + '../../services/graph-run-sse',
  26 + '../../services/graph-engine',
  27 + '../../routes/graph-run',
  28 +];
  29 +
  30 +describe('POST /api/graph-runs/:id/nodes/:nodeId/force-fail', () => {
  31 + let wdb;
  32 + let srv;
  33 + let base;
  34 + let runsDir;
  35 + let seq = 0;
  36 + // 每个用例一个独立 company：一家公司同时只有一条活跃 root run（run-slot 约束），
  37 + // 同一个 company 起第二条 run 会撞 RunSlotBusyError。
  38 + const nextCompany = () => `co-force-fail-${++seq}`;
  39 +
  40 + beforeAll(async () => {
  41 + for (const f of DB_FILES) { try { fs.unlinkSync(f); } catch {} }
  42 + process.env.GRAPH_DB_PATH = TEST_DB_PATH;
  43 + runsDir = fs.mkdtempSync(path.join(os.tmpdir(), 'force-fail-runs-'));
  44 + process.env.GRAPH_RUNS_DIR = runsDir;
  45 + for (const m of MODULES) delete require.cache[require.resolve(m)];
  46 + wdb = require('../../services/work-graph-db');
  47 +
  48 + const app = express();
  49 + app.use(express.json());
  50 + app.use('/api/graph-runs', require('../../routes/graph-run'));
  51 + await new Promise((resolve) => {
  52 + srv = app.listen(0, '127.0.0.1', () => {
  53 + base = `http://127.0.0.1:${srv.address().port}`;
  54 + resolve();
  55 + });
  56 + });
  57 + });
  58 +
  59 + afterAll(async () => {
  60 + if (srv) await new Promise((r) => srv.close(r));
  61 + const { removeTestCompany } = require('../helpers/ensure-test-company');
  62 + for (let i = 1; i <= seq; i++) removeTestCompany(`co-force-fail-${i}`);
  63 + for (const f of DB_FILES) { try { fs.unlinkSync(f); } catch {} }
  64 + delete process.env.GRAPH_DB_PATH;
  65 + try { fs.rmSync(runsDir, { recursive: true, force: true }); } catch {}
  66 + delete process.env.GRAPH_RUNS_DIR;
  67 + for (const m of MODULES) delete require.cache[require.resolve(m)];
  68 + });
  69 +
  70 + function ensureAgents(ids) {
  71 + const db = wdb.getDb();
  72 + const now = new Date().toISOString();
  73 + const ins = db.prepare(`
  74 + INSERT OR IGNORE INTO system_agents
  75 + (id, name, scope, owner_company_id, legacy_path, status, created_at, updated_at)
  76 + VALUES (?, ?, 'public', NULL, ?, 'active', ?, ?)
  77 + `);
  78 + for (const id of new Set(ids)) ins.run(id, id, `system/agents/${id}`, now, now);
  79 + }
  80 +
  81 + /** 建一条只有一个节点的 work run，直接把节点摁成指定状态（不关心怎么走到这个状态，
  82 + * 只关心端点在这个状态下的行为——执行语义已经在别的测试文件覆盖过）。 */
  83 + function buildRunWithNodeStatus(status) {
  84 + const companyId = nextCompany();
  85 + require('../helpers/ensure-test-company').ensureTestCompany(companyId);
  86 + ensureAgents(['w']);
  87 + const g = wdb.createWorkGraph({ companyId, nodes: ['w'], edges: [] });
  88 + expect(g.errors || []).toEqual([]);
  89 + const run = wdb.createRun({ graphKind: 'work', graphId: g.id, companyId, executionMode: 'manual' });
  90 + require('../../services/graph-engine').markRunReady(run.id);
  91 + const node = wdb.getRunState(run.id).nodes[0];
  92 + const now = new Date().toISOString();
  93 + wdb.getDb()
  94 + .prepare(`UPDATE graph_run_nodes SET status = ?, started_at = ?, updated_at = ? WHERE id = ?`)
  95 + .run(status, now, now, node.id);
  96 + return { runId: run.id, nodeRunId: node.id };
  97 + }
  98 +
  99 + it('running 节点调用成功，转 fail 后交给既有失败路由（无边可走 → blocked，可被 retry 接上）', async () => {
  100 + const { runId, nodeRunId } = buildRunWithNodeStatus('running');
  101 + const r = await fetch(`${base}/api/graph-runs/${runId}/nodes/${nodeRunId}/force-fail`, {
  102 + method: 'POST',
  103 + headers: { 'Content-Type': 'application/json' },
  104 + body: JSON.stringify({}),
  105 + });
  106 + expect(r.status).toBe(200);
  107 + const body = await r.json();
  108 + expect(body.ok).toBe(true);
  109 + // 单节点图没有 fail 边可走 → blockNode，落在人工可 retry 的终点
  110 + expect(body.nodeStatus).toBe('blocked');
  111 +
  112 + // 验证确实接得上 retry 端点（这才是「可被 retry 端点接上」这条验收的真实含义）
  113 + const retryRes = await fetch(`${base}/api/graph-runs/${runId}/nodes/${nodeRunId}/retry`, { method: 'POST' });
  114 + expect(retryRes.status).toBe(200);
  115 + const retryBody = await retryRes.json();
  116 + expect(retryBody.ok).toBe(true);
  117 + expect(retryBody.nodeStatus).toBe('ready');
  118 + });
  119 +
  120 + it('自定义 failReason 会原样落进 fail_reason（走常规选边，不是 infra_flake 特判）', async () => {
  121 + const { runId, nodeRunId } = buildRunWithNodeStatus('running');
  122 + const r = await fetch(`${base}/api/graph-runs/${runId}/nodes/${nodeRunId}/force-fail`, {
  123 + method: 'POST',
  124 + headers: { 'Content-Type': 'application/json' },
  125 + body: JSON.stringify({ failReason: 'stuck_no_progress' }),
  126 + });
  127 + expect(r.status).toBe(200);
  128 + // 单节点图没有 fail 边，无论什么 fail_reason 都落 blocked——这里只确认端点把自定义原因
  129 + // 传下去了没炸，选边逻辑本身已经在别处测过
  130 + const body = await r.json();
  131 + expect(body.ok).toBe(true);
  132 + expect(body.nodeStatus).toBe('blocked');
  133 + });
  134 +
  135 + for (const status of ['fail', 'blocked', 'pending', 'ready', 'success']) {
  136 + it(`节点 status='${status}' 时调用返回 400（不是 running 就不该走这个端点）`, async () => {
  137 + const { runId, nodeRunId } = buildRunWithNodeStatus(status);
  138 + const r = await fetch(`${base}/api/graph-runs/${runId}/nodes/${nodeRunId}/force-fail`, {
  139 + method: 'POST',
  140 + headers: { 'Content-Type': 'application/json' },
  141 + body: JSON.stringify({}),
  142 + });
  143 + expect(r.status).toBe(400);
  144 + const body = await r.json();
  145 + expect(body.op).toBe('force-fail');
  146 + expect(body.nodeStatus).toBe(status);
  147 + expect(body.allowed).toEqual(['running']);
  148 + });
  149 + }
  150 +
  151 + it('节点不存在返回 404', async () => {
  152 + const { runId } = buildRunWithNodeStatus('running');
  153 + const r = await fetch(`${base}/api/graph-runs/${runId}/nodes/nope/force-fail`, {
  154 + method: 'POST',
  155 + headers: { 'Content-Type': 'application/json' },
  156 + body: JSON.stringify({}),
  157 + });
  158 + expect(r.status).toBe(404);
  159 + });
  160 +});
... ...