宜博 / lemonboss

/**
 * Unit Tests — 子 run 转终态回写父节点（27-P3B 界面走查发现的永久卡死）
 *
 * 界面走查（Claude in Chrome 真点）实测的复现路径：
 *   org 节点的子 work run 第一次失败 → 父 org 节点判 fail → 入口节点没有 fail 出边 → blocked
 *   → 用户在中栏把子 run 修好、跑到 completed
 *   → **父节点还钉在 blocked，root run 也钉在 blocked，org 层永远推不到下一个节点**
 *
 * 根因：propagateToParent 的幂等守卫写的是
 *     if (parentNode.status !== 'running') return;   // 「已经结算过 → 幂等返回」
 * 判据用错了维度 —— 它想防「同一个结果被结算两次」，实际拦掉的是「子 run 换了个结果」。
 * subgraph 节点的状态本来就是从子 run 派生的，子 run 换了终态就该重新派生。
 *
 * 当时唯一的人工出路是对 org 节点点「重试」，而那会把整张子图重置到下一个 attempt，
 * 刚跑完的成果全丢 —— 所以这条必须进 CI，不能只靠人点界面才发现。
 *
 * 隔离手法同 graph-engine-node-error.test.js：先设 GRAPH_DB_PATH / GRAPH_RUNS_DIR 再 require，
 * 仓库里的 data/graph.sqlite 与 data/companies/ 全程不被碰到；executor 全是注入的假货。
 */
import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
import fs from 'fs';
import os from 'os';
import path from 'path';

describe('子 run 转终态回写父节点', () => {
  let wdb;
  let engine;
  let sse;
  let runsDir;
  const TEST_DB_PATH = '/tmp/test-graph-engine-revive-parent.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-revive-runs-'));
    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');
    sse = require('../../services/graph-run-sse');
    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);
    engine.setPlanWorkGraph(null);
  });

  let seq = 0;
  const nextCompany = () => `revive-co-${++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, source_kind, legacy_path, status, created_at, updated_at)
      VALUES (?, ?, 'public', NULL, 'global_agent', ?, 'active', ?, ?)
    `);
    for (const id of new Set(ids)) ins.run(id, id, `system/agents/${id}`, now, now);
  }

  const nodeByName = (runId, name) => {
    const st = wdb.getRunState(runId);
    return (st.nodes || []).find((n) => n.name === name);
  };

  /**
   * 造出走查里那个局面：父 run 一个 subgraph 节点 + 一张会失败的子图。
   * 父图的入口节点故意**没有 fail 出边** —— 走查里就是这么走到 blocked 的。
   */
  function buildParentWithFailingChild() {
    const companyId = nextCompany();
    ensureAgents(['org-planner', 'wg-only']);

    const childG = wdb.createWorkGraph({ companyId, nodes: ['wg-only'], edges: [] });
    expect(childG.errors || []).toEqual([]);

    const parentG = wdb.createWorkGraph({
      companyId,
      nodes: [{ nodeName: 'org-planner', systemAgentId: 'org-planner', nodeKind: 'subgraph' }],
      edges: [],
    });
    expect(parentG.errors || []).toEqual([]);

    const parentRun = wdb.createRun({
      graphKind: 'work', graphId: parentG.id, companyId, executionMode: 'manual',
    });
    engine.markRunReady(parentRun.id);
    engine.startRun(parentRun.id, { mode: 'manual' });   // advanceOne 要求 run_status='running'

    // 子图规划注入：第一次执行该 subgraph 节点时把上面那张 childG 挂上去
    engine.setPlanWorkGraph(() => ({ workGraphId: childG.id }));
    // 子图里那个节点固定失败，且不给 fail_reason（走查里就是 fail_reason=null）
    engine.setAgentExecutor(async () => ({ status: 'fail', failReason: null, output: '子节点失败' }));

    return { companyId, parentRunId: parentRun.id };
  }

  /** 把父 run 推到「子 run 失败、父节点 blocked」的状态 */
  async function driveToBlocked(parentRunId) {
    for (let i = 0; i < 12; i += 1) {
      const r = await engine.advanceTree(parentRunId);
      if (!r || r.ok === false) break;
    }
    const parent = nodeByName(parentRunId, 'org-planner');
    return parent;
  }

  it('复现前提：子 run 失败会把父 subgraph 节点推到 blocked', async () => {
    const { parentRunId } = buildParentWithFailingChild();
    const parent = await driveToBlocked(parentRunId);

    expect(parent.status).toBe('blocked');
    expect(parent.childRunId).toBeTruthy();
    expect(wdb.getRunState(parentRunId).run.run_status).toBe('blocked');
  });

  it('子 run 后来跑到 completed 时，父节点从 blocked 被重新派生为 success（不再永久卡死）', async () => {
    const { parentRunId } = buildParentWithFailingChild();
    const blocked = await driveToBlocked(parentRunId);
    const childRunId = blocked.childRunId;
    expect(blocked.status).toBe('blocked');

    // 用户在中栏把子 run 修好：换成会成功的执行器，重试子图里那个失败节点并跑完
    engine.setAgentExecutor(async () => ({ status: 'success', output: '修好了' }));
    const childNode = wdb.getRunState(childRunId).nodes[0];
    engine.retryNode(childRunId, childNode.id);
    for (let i = 0; i < 6; i += 1) {
      const r = await engine.advanceOne(childRunId);
      if (!r || r.ok === false) break;
    }
    expect(wdb.getRunState(childRunId).run.run_status).toBe('completed');

    // 关键断言：父节点不再钉在 blocked
    const parent = nodeByName(parentRunId, 'org-planner');
    expect(parent.status).toBe('success');
    expect(wdb.getRunState(parentRunId).run.run_status).toBe('completed');
  });

  it('复活走的是 blocked → ready → running → success 三步合法转换，timeline 上留得下痕迹', async () => {
    const { parentRunId } = buildParentWithFailingChild();
    const blocked = await driveToBlocked(parentRunId);
    const childRunId = blocked.childRunId;

    engine.setAgentExecutor(async () => ({ status: 'success', output: '修好了' }));
    const childNode = wdb.getRunState(childRunId).nodes[0];
    engine.retryNode(childRunId, childNode.id);
    for (let i = 0; i < 6; i += 1) {
      const r = await engine.advanceOne(childRunId);
      if (!r || r.ok === false) break;
    }

    const lines = fs.readFileSync(sse.timelinePath(parentRunId), 'utf8')
      .split('\n').filter(Boolean).map((l) => JSON.parse(l));
    const revived = lines.filter((l) => l.payload && l.payload.revived_from_child === childRunId);
    const hop = (l) => `${l.payload.from}→${l.payload.to}`;

    // 节点轨：blocked → ready → running 两跳都要在，否则说明是靠放宽转换表硬转过去的
    expect(revived.filter((l) => l.kind === 'node.status').map(hop))
      .toEqual(['blocked→ready', 'ready→running']);
    // run 轨：父 run 也得被拉回 running，否则节点复活了 run 还钉在 blocked
    //（RUN_TRANSITIONS 没有 blocked → completed，settleRun 在转换非法时是静默跳过的）
    expect(revived.filter((l) => l.kind === 'run.status').map(hop)).toEqual(['blocked→running']);
  });

  it('子 run 再失败一轮时父节点不动（只认转好，不追加惩罚）', async () => {
    // propagateToParent 不导出（本模块的约定：不为测试开公开面），
    // 所以这里用真入口触发第二次回写：重试子节点、让它再失败一次，
    // 子 run 重新收敛到 blocked 的路上引擎自己会调 propagateToParent。
    const { parentRunId } = buildParentWithFailingChild();
    const blocked = await driveToBlocked(parentRunId);
    const childRunId = blocked.childRunId;

    const childNode = wdb.getRunState(childRunId).nodes[0];
    engine.retryNode(childRunId, childNode.id);
    for (let i = 0; i < 6; i += 1) {
      const r = await engine.advanceOne(childRunId);
      if (!r || r.ok === false) break;
    }
    expect(wdb.getRunState(childRunId).run.run_status).toBe('blocked');

    const parent = nodeByName(parentRunId, 'org-planner');
    expect(parent.status).toBe('blocked');
  });
});