graph-engine-stop-run-orphan-race.test.js
8.05 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
/**
* Unit Test — stopRun 不能给「正在 executeSubgraph 里 await planWorkGraph」的节点留孤儿子 run
* （T-mtgxcnfz 复现）
*
* 真实事故时序：run_e4ba218e9446 于 2026-08-31T07:30:50.644Z 落 stopped，它的子 run
* run_12aa199b07f2 created_at 是 07:30:50.607Z（早 37ms，同一次事件循环附近），但从未被置
* stop_requested，一直跑到人工手动 /stop 才收掉。
*
* 根因：`stopRun`（graph-engine.js）在函数开头一次性拍 `descendantRuns(run)` 快照决定要停
* 哪些 run；`executeSubgraph`（同文件）建子 run 之前有一次真实 `await planWorkGraph(ctx)`——
* 如果这次 await 跨越了 stopRun 执行的时间点，子 run 在 stopRun 拍完快照**之后**才落库，
* 永远不在 targets 里，`stopRun` 本身又是全同步函数、没有内部 await，事后补扫赶不上。
*
* 修法钉在 executeSubgraph：`await planWorkGraph` 之后、真正 `createRun` 之前重新读一次
* 父 run，`stop_requested` 已经变真就放弃创建子 run，就地按节点失败结算——不产生孤儿。
*
* 用真实的 org run + subgraph 节点复现（不是打桩内部实现细节）：`setPlanWorkGraph` 注入一个
* 可以从测试里手动控制何时 resolve 的假规划，制造出与生产一致的 await 窗口，在窗口内调用
* `engine.stopRun`，再放行规划，断言子 run 没有被建出来、父节点按 fail 收敛、run 落 stopped。
*/
import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
import fs from 'fs';
import os from 'os';
import path from 'path';
describe('stopRun 与 executeSubgraph 的建子 run 竞态（T-mtgxcnfz）', () => {
let orgSvc;
let wdb;
let engine;
let runsDir;
const TEST_DB_PATH = '/tmp/test-graph-engine-stop-run-orphan-race.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-stop-run-orphan-race-'));
process.env.GRAPH_RUNS_DIR = runsDir;
for (const m of [
'../../services/graph-db',
'../../services/org-graph-service',
'../../services/work-graph-db',
'../../services/graph-run-sse',
'../../services/graph-engine',
]) {
delete require.cache[require.resolve(m)];
}
orgSvc = require('../../services/org-graph-service');
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 org_graph_edges;
DELETE FROM org_graph_nodes;
DELETE FROM system_agents;
`);
db.pragma('foreign_keys = ON');
engine.setAgentExecutor(null);
engine.setPlanWorkGraph(null);
});
let seq = 0;
const nextCompany = () => `co-stop-race-${++seq}`;
it('planWorkGraph 还没 resolve 时被 stopRun 停掉：不建子 run，节点按 fail 收敛，run 落 stopped', async () => {
const companyId = nextCompany();
expect(orgSvc.writeGraph(companyId, { nodes: ['org-coder'], edges: [] }).ok).toBe(true);
let releasePlanning;
const planningStarted = new Promise((resolveStarted) => {
engine.setPlanWorkGraph((ctx) => {
resolveStarted();
return new Promise((resolve) => {
releasePlanning = () => {
// 造一张真实可用的子图定义——workGraphId 不能是 null，否则会在
// executeSubgraph 里被「未返回 workGraphId」那道**无关**的校验拦下，
// 那不是这条测试要盯的路径，会把「没建成子 run」的原因张冠李戴。
const worker = `${ctx.nodeName}-worker`;
const db = wdb.getDb();
const now = new Date().toISOString();
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', ?, ?)
`).run(worker, worker, `system/agents/${worker}`, now, now);
const g = wdb.createWorkGraph({ companyId, nodes: [worker], edges: [] });
resolve({ workGraphId: g.id });
};
});
});
});
const run = wdb.createRun({ graphKind: 'org', graphId: companyId, companyId, executionMode: 'manual' });
engine.markRunReady(run.id);
engine.startRun(run.id, { mode: 'manual' });
// 不 await：advanceOne 内部会在 executeSubgraph 里卡在 planWorkGraph 那次 await 上，
// 这条 promise 要等我们手动 releasePlanning 之后才会继续往下走。
const stepPromise = engine.nextStep(run.id);
await planningStarted; // 确认已经真的进了 await 窗口，不是还没排到
// 窗口期间人工调 /stop——此时子 run 还没建出来，descendantRuns(run) 快照里只有 run 自己。
const stopResult = engine.stopRun(run.id);
expect(stopResult.ok).toBe(true);
expect(wdb.getRun(run.id).run_status).toBe('stopped');
// 这一刻整棵树只有 root 自己——旧代码的 bug 正是「这个断言此刻成立，但之后会被打破」
expect(wdb.listRunTree(run.id)).toHaveLength(1);
// 放行规划，让 executeSubgraph 走到「要不要建子 run」那一步
releasePlanning();
const stepResult = await stepPromise;
expect(stepResult.ok).toBe(true);
// 核心断言：没有孤儿子 run——树上除了 root 一个都没多出来
const tree = wdb.listRunTree(run.id);
expect(tree).toHaveLength(1);
// 节点没有 pending 的 child_run_id，且不再停在 running 干等一个不存在的子 run——
// 没有匹配的 fail 边时 routeFailure 落 blocked（需要人工介入），这里不咬死具体是
// fail 还是 blocked，只钉住「不是孤儿」这条：没有子 run、没有继续假装在跑。
const node = wdb.getRunState(run.id).nodes.find((n) => n.name === 'org-coder');
expect(node.childRunId).toBeNull();
expect(node.status).not.toBe('running');
expect(['fail', 'blocked']).toContain(node.status);
});
it('对照组：没有并发 stop 时，同样的 await 窗口结束后子 run 正常建出来', async () => {
const companyId = nextCompany();
expect(orgSvc.writeGraph(companyId, { nodes: ['org-coder'], edges: [] }).ok).toBe(true);
engine.setPlanWorkGraph(async (ctx) => {
const worker = `${ctx.nodeName}-worker`;
const db = wdb.getDb();
const now = new Date().toISOString();
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', ?, ?)
`).run(worker, worker, `system/agents/${worker}`, now, now);
const g = wdb.createWorkGraph({ companyId, nodes: [worker], edges: [] });
expect(g.errors || []).toEqual([]);
return { workGraphId: g.id };
});
const run = wdb.createRun({ graphKind: 'org', graphId: companyId, companyId, executionMode: 'manual' });
engine.markRunReady(run.id);
engine.startRun(run.id, { mode: 'manual' });
const stepResult = await engine.nextStep(run.id);
expect(stepResult.ok).toBe(true);
expect(wdb.listRunTree(run.id)).toHaveLength(2);
const node = wdb.getRunState(run.id).nodes.find((n) => n.name === 'org-coder');
expect(node.childRunId).toBeTruthy();
expect(node.status).toBe('running'); // 等子 run 收敛，父节点停在 running
});
});