宜博 / lemonboss

/**
 * 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: [] });
  });
});