/** * Unit Tests — 恢复路径 × 执行坑(27-P3C 批 3 对抗复核 F1 / F2) * * 三条恢复转换(人加完预算点启动 / 子 run 救回父 run / 对已收敛 run 里的节点点重试)的 from * 全在活跃集之外,于是全部命中 `updateRunStatus` 的准入守卫;而批 3 让「坑一空立刻被队里 * 下一条填上」成了必然事件。这个文件钉死修完之后的三件事: * 1. 人发起的恢复撞上坑满 → **退回队列**,坑一空调度器自己把它接回来(不再永远躺在 blocked); * 2. 退不回队列的(没有 task 的裸 run)→ **干净拒绝**:抛 RunSlotBusyError 时节点一个字都没写; * 3. `syncRunNodesToDef` 是白名单:只碰 queued 的 run,正在跑的 run 一行都不许删。 * * 隔离手法与 task-scheduler.test.js 一致:先设 GRAPH_DB_PATH / GRAPH_RUNS_DIR 再 require。 */ import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; import fs from 'fs'; import os from 'os'; import path from 'path'; describe('恢复路径与执行坑', () => { let wdb; let engine; let sse; let scheduler; let runsDir; const TEST_DB_PATH = '/tmp/test-slot-recovery.sqlite'; const pending = []; 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(), 'slot-recovery-runs-')); 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', '../../services/task-settle', '../../services/task-scheduler', ]) { delete require.cache[require.resolve(m)]; } wdb = require('../../services/work-graph-db'); sse = require('../../services/graph-run-sse'); engine = require('../../services/graph-engine'); scheduler = require('../../services/task-scheduler'); }); /** advanceOne 的 ensureProjectRoot 认的是仓库里的 data/companies/,只删本文件造的那些 */ function cleanupLeakedCompanyDirs() { const root = path.resolve(__dirname, '../../../data/companies'); if (!fs.existsSync(root)) return; for (const cid of fs.readdirSync(root)) { if (!cid.startsWith('slotrec-co-')) continue; const dir = path.join(root, cid); if (fs.existsSync(path.join(dir, 'company.json'))) continue; // 那是真公司 try { fs.rmSync(dir, { recursive: true, force: true }); } catch { /* 删不掉就留着 */ } } } afterAll(async () => { for (const p of pending) p(); await new Promise((r) => setTimeout(r, 100)); scheduler.stopSweeper(); cleanupLeakedCompanyDirs(); 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 tasks; DELETE FROM goals; 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 = () => `slotrec-co-${++seq}`; function ensureAgents(ids) { const now = new Date().toISOString(); const ins = wdb.getDb().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 buildGraph(companyId, nodes = ['solo']) { ensureAgents(nodes); const edges = nodes.slice(1).map((to, i) => ({ from: nodes[i], to })); const g = wdb.createWorkGraph({ companyId, nodes, edges }); expect(g.errors || []).toEqual([]); return g.id; } /** 入队三步(端点将来做的事):建 task → 建 queued run → task 挂 current_run_id */ function enqueue(companyId, goalId, { title, priority = 'P1', graphId } = {}) { const task = wdb.createTask({ goalId, companyId, title, description: `${title} 的描述`, executionMode: 'auto', priority, }); const run = wdb.createRun({ graphKind: 'work', graphId: graphId || buildGraph(companyId), companyId, goalId, taskId: task.id, executionMode: 'auto', runStatus: 'queued', }); wdb.updateTask(task.id, { status: 'queued', currentRunId: run.id, queuedAt: new Date().toISOString() }); return { taskId: task.id, runId: run.id }; } const newGoal = (companyId) => wdb.createGoal({ companyId, title: 'g', description: 'g' }).id; /** 永不 resolve 的 executor:run 稳稳停在 running,坑一直被占着 */ function hangingExecutor() { engine.setAgentExecutor(() => new Promise((resolve) => pending.push(() => resolve({ status: 'success' })))); } const flush = () => new Promise((r) => setImmediate(() => setImmediate(r))); // ──────────────────────────────────────────────────────────── // F1 场景 A:预算耗尽 → blocked → 坑被 T2 顶走 → 人加完预算点启动 // ──────────────────────────────────────────────────────────── it('恢复撞上坑满时退回队列,坑一空由调度器接回来(T1 不会永远躺在 blocked)', async () => { const companyId = nextCompany(); const goalId = newGoal(companyId); hangingExecutor(); const t1 = enqueue(companyId, goalId, { title: 't1', priority: 'P0' }); const t2 = enqueue(companyId, goalId, { title: 't2' }); scheduler.tick(companyId); expect(wdb.getRun(t1.runId).run_status).toBe('running'); // 预算耗尽那一跳:run → blocked。漏斗钩子把坑让给 T2(setImmediate 出栈后才跑) expect(sse.transitionRun(t1.runId, 'running', 'blocked', {}, { summary: '预算耗尽' })).toBeTruthy(); await flush(); expect(wdb.getRun(t2.runId).run_status).toBe('running'); expect(wdb.getTask(t1.taskId).status).toBe('blocked'); // 人加完预算点「启动」:坑已经不在了 —— 不许 500,也不许原地卡死,而是回队列 const res = engine.startRun(t1.runId, { mode: 'auto' }); expect(res).toMatchObject({ ok: true, requeued: true, runStatus: 'queued' }); expect(wdb.getRun(t1.runId).run_status).toBe('queued'); expect(wdb.getTask(t1.taskId).status).toBe('queued'); expect(wdb.getTask(t1.taskId).blocked_reason).toBeNull(); // T2 收敛 → 坑空 → T1 按位次(P0)被接回来。全程无人再点一次按钮。 expect(sse.transitionRun(t2.runId, 'running', 'completed', {}, { summary: '用例直推终态' })).toBeTruthy(); await flush(); expect(wdb.getRun(t1.runId).run_status).toBe('running'); expect(wdb.getTask(t1.taskId).status).toBe('running'); }); // ──────────────────────────────────────────────────────────── // F1 场景 B:retryNode 的 run 转换排在节点转换之前 // ──────────────────────────────────────────────────────────── it('坑满且退不回队列时 retryNode 干净拒绝:节点状态与 attempt 一个字都没动', async () => { const companyId = nextCompany(); const goalId = newGoal(companyId); hangingExecutor(); const t1 = enqueue(companyId, goalId, { title: 't1' }); scheduler.tick(companyId); expect(wdb.getRun(t1.runId).run_status).toBe('running'); // 一条**没有 task** 的已收敛 run(裸 run 回不了队列 —— 队列是按 task 选候选的) const db = wdb.getDb(); const other = wdb.createRun({ graphKind: 'work', graphId: buildGraph(companyId, ['x1']), companyId }); db.prepare(`UPDATE graph_runs SET run_status = 'completed' WHERE id = ?`).run(other.id); const node = wdb.getRunState(other.id).nodes[0]; db.prepare(`UPDATE graph_run_nodes SET status = 'fail' WHERE id = ?`).run(node.id); expect(() => engine.retryNode(other.id, node.id)).toThrow(wdb.RunSlotBusyError); // 半生效才是真正的杀手:节点 ready + attempt+1 而 run 还是 completed 的话, // advanceTree 只捡 running 的 run,这个节点永远不会被推,而且没有任何路径能修回来。 const after = wdb.getRunNode(node.id); expect(after.status).toBe('fail'); expect(after.attempt).toBe(0); expect(wdb.getRun(other.id).run_status).toBe('completed'); await flush(); }); it('RunSlotBusyError 带得出「谁占着坑」,路由层才翻得成一句人能读的 409', () => { const companyId = nextCompany(); const goalId = newGoal(companyId); hangingExecutor(); const t1 = enqueue(companyId, goalId, { title: 't1' }); scheduler.tick(companyId); const other = wdb.createRun({ graphKind: 'work', graphId: buildGraph(companyId, ['y1']), companyId }); let caught = null; try { wdb.updateRunStatus(other.id, 'planning', 'ready'); } catch (err) { caught = err; } expect(caught).toBeInstanceOf(wdb.RunSlotBusyError); expect(caught.companyId).toBe(companyId); expect(caught.heldBy).toEqual([t1.runId]); // 这里以前断言的是「message 里带 constraint」,理由写着「批 4 之前 routes/company.js // 还靠 /UNIQUE|constraint/i 嗅探这句文本翻 409」—— 那两处 catch 批 3 就删干净了, // 断言一句没人读的报文等于给报错文案上了一把没有钥匙的锁。 // 真正要钉的是「调用方认得出它」:判据只有 work-graph-db.isSlotRaceError 一处,按类型判。 expect(wdb.isSlotRaceError(caught)).toBe(true); // 报文仍要点得出「是哪家公司的坑、被谁占着」,那是给人读的,不是给正则读的 expect(caught.message).toContain(companyId); expect(caught.message).toContain(t1.runId); }); // ──────────────────────────────────────────────────────────── // F2:syncRunNodesToDef 是白名单 // ──────────────────────────────────────────────────────────── it('syncRunNodesToDef 只碰 queued 的 run:正在跑的 run 一行都不许删', () => { const companyId = nextCompany(); const goalId = newGoal(companyId); hangingExecutor(); const graphId = buildGraph(companyId, ['n1', 'n2']); const t1 = enqueue(companyId, goalId, { title: 't1', graphId }); scheduler.tick(companyId); expect(wdb.getRun(t1.runId).run_status).toBe('running'); // 「重新规划组织」:定义里删掉 n2(真删行 —— def_node_id 认的是节点行 id) const db = wdb.getDb(); db.prepare(`DELETE FROM work_graph_edges WHERE work_graph_id = ?`).run(graphId); db.prepare(`DELETE FROM work_graph_nodes WHERE work_graph_id = ? AND node_name = 'n2'`).run(graphId); // 多进程下这一句会与「另一个进程已经把 run 推成 running 并开跑」交叠: // 黑名单放行 running,于是 n2 的行被直接 DELETE,run 少跑一段图还判 completed。 expect(wdb.syncRunNodesToDef(t1.runId)).toBeNull(); expect(wdb.getRunState(t1.runId).nodes.length).toBe(2); }); it('退回队列的 run 再出队时,跑过痕迹的孤儿节点留着不删(pending 不等于没跑过)', () => { const companyId = nextCompany(); const goalId = newGoal(companyId); const graphId = buildGraph(companyId, ['m1', 'm2']); const t1 = enqueue(companyId, goalId, { title: 't1', graphId }); // 回退重置会把**跑过的**节点写回 pending 并保留 attempt —— 这正是「只删 pending 是安全的」 // 那句旧注释不成立的地方。这里直接造出那种行。 const db = wdb.getDb(); const m2 = wdb.getRunState(t1.runId).nodes.find((n) => n.name === 'm2'); db.prepare(`UPDATE graph_run_nodes SET status = 'pending', attempt = 2 WHERE id = ?`).run(m2.id); db.prepare(`DELETE FROM work_graph_edges WHERE work_graph_id = ?`).run(graphId); db.prepare(`DELETE FROM work_graph_nodes WHERE work_graph_id = ? AND node_name = 'm2'`).run(graphId); const diff = wdb.syncRunNodesToDef(t1.runId); expect(diff.removed).toEqual([]); expect(diff.kept.length).toBe(1); expect(wdb.getRunNode(m2.id)).toBeTruthy(); // 审计轨迹还在 }); });