/** * Unit Tests — graph-engine 的持续性存活扫描(45 号文档 Part A) * * 场景:服务器主进程没有重启,但驱动某个 agent_call 节点的 Claude CLI 会话在调用中途 * 死掉/被打断,`graph_run_nodes.status` 永远停在 'running'。开机一次性孤儿扫描 * (`sweepOrphanedRunningNodes`,按「早于本次进程启动时间」判定)看不到这种情况——本文件 * 测的是它的持续性版本:`sweepStaleRunningAgentCalls`,按「running 超过一个阈值」判定, * 且只扫 `node_kind='agent_call'`(subgraph 节点的"运行时长"横跨多个内部调用,不适用 * 同一固定阈值,本次不做)。 * * 隔离手法与 graph-engine-orphan-sweep.test.js 一致。 */ 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-stale-agent-call-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-stale-agent-call-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-stale-${++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); function markAsRunning(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; } const NOW = '2026-08-31T10:00:00.000Z'; it('agent_call 节点 running 超过阈值判定为存活超时,走 infra_flake 原地重跑', () => { const companyId = nextCompany(); const { runId } = buildWorkRun({ companyId, nodes: ['w'], edges: [] }); // 默认阈值 25 分钟;30 分钟前开始 = 超阈值 markAsRunning(runId, 'w', '2026-08-31T09:30:00.000Z'); const result = engine.sweepStaleRunningAgentCalls({ nowIso: NOW }); expect(result.scanned).toBe(1); expect(result.swept).toHaveLength(1); expect(result.swept[0]).toMatchObject({ runId, infraRerun: true }); const fresh = nodeByName(runId, 'w'); expect(fresh.status).toBe('ready'); expect(fresh.attempt).toBe(0); }); it('running 未超过阈值的节点不受影响(大概率还在真实跑)', () => { const companyId = nextCompany(); const { runId } = buildWorkRun({ companyId, nodes: ['w'], edges: [] }); // 5 分钟前开始,远低于默认 25 分钟阈值 markAsRunning(runId, 'w', '2026-08-31T09:55:00.000Z'); const result = engine.sweepStaleRunningAgentCalls({ nowIso: NOW }); expect(result.scanned).toBe(0); expect(nodeByName(runId, 'w').status).toBe('running'); }); it('阈值可调:传更短的 thresholdMs 会让原本"没超时"的节点也被扫到', () => { const companyId = nextCompany(); const { runId } = buildWorkRun({ companyId, nodes: ['w'], edges: [] }); markAsRunning(runId, 'w', '2026-08-31T09:55:00.000Z'); // 5 分钟前 const result = engine.sweepStaleRunningAgentCalls({ nowIso: NOW, thresholdMs: 60 * 1000 }); // 阈值 1 分钟 expect(result.scanned).toBe(1); expect(nodeByName(runId, 'w').status).toBe('ready'); }); it('subgraph 节点不在扫描范围内,即使 running 很久也不受影响', () => { const companyId = nextCompany(); ensureAgents(['s']); const g = wdb.createWorkGraph({ companyId, nodes: [{ nodeName: 's', nodeKind: 'subgraph' }], edges: [] }); expect(g.errors || []).toEqual([]); const run = wdb.createRun({ graphKind: 'work', graphId: g.id, companyId, executionMode: 'manual' }); engine.markRunReady(run.id); markAsRunning(run.id, 's', '2026-08-31T08:00:00.000Z'); // 2 小时前,远超阈值 const result = engine.sweepStaleRunningAgentCalls({ nowIso: NOW }); expect(result.scanned).toBe(0); expect(nodeByName(run.id, 's').status).toBe('running'); }); it('没有任何超时节点时是安静的 no-op', () => { const companyId = nextCompany(); buildWorkRun({ companyId, nodes: ['w'], edges: [] }); // 节点是 ready,不是 running const result = engine.sweepStaleRunningAgentCalls({ nowIso: NOW }); expect(result).toEqual({ scanned: 0, swept: [] }); }); });