宜博 / lemonboss

/**
 * Unit Tests — 节点失败时原因留不留得下来（27-P3B W12，「第 12 处四环漏一环」）
 *
 * 修之前：advanceOne 的 executor catch 把 err.message 写进 outcome.output，
 * 而 outcome.output **一个消费方都没有** —— 不落库（graph_run_nodes 没有 output 列）、
 * 不进 SSE、也不写 timeline，fail_reason 还是 null。真跑第一个 work 节点时
 * AgentManager 抛的「Agent not found」就是这么消失的，操作者看到「节点 fail、无任何原因」。
 *
 * 这里锁的是修完之后的四环：
 *   产生（catch / executor 返回值）→ 传递（timelineNote / node.output 事件 data）
 *   → 落库（timeline.ndjson + stream.log，append-only）→ 读回（本用例就是按读回的那条路读的：
 *   前端走 GET /api/graph-runs/:id/stream?lastEventId=0 全量重放同一份 stream.log）
 *
 * 隔离手法同 graph-engine.test.js：先设 GRAPH_DB_PATH / GRAPH_RUNS_DIR 再 require，
 * 仓库里的 data/graph.sqlite 与 data/companies/ 全程不被碰到；executor 全是注入的假货，
 * 不起任何 Claude CLI。
 */
import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
import fs from 'fs';
import os from 'os';
import path from 'path';

describe('graph-engine 节点失败留痕（W12）', () => {
  let wdb;
  let engine;
  let sse;
  let runsDir;
  const TEST_DB_PATH = '/tmp/test-graph-engine-node-error.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-node-error-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);
  });

  let seq = 0;
  const nextCompany = () => `w12-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);
  }

  function buildRun(nodes, edges = []) {
    const companyId = nextCompany();
    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 run.id;
  }

  /** 读回轨 1：审计轨 timeline.ndjson（append-only） */
  const timeline = (runId) =>
    fs.readFileSync(sse.timelinePath(runId), 'utf8')
      .split('\n').filter(Boolean).map((l) => JSON.parse(l));

  /** 读回轨 2：stream.log —— 前端 lastEventId=0 全量重放读的就是这一份 */
  const streamEvents = (runId) =>
    fs.readFileSync(path.join(runsDir, runId, 'stream.log'), 'utf8')
      .split('\n\n').filter(Boolean)
      .map((block) => {
        const ev = /^event: (.+)$/m.exec(block);
        const data = /^data: (.+)$/m.exec(block);
        return { event: ev && ev[1], data: JSON.parse(data[1]) };
      });

  const nodeByName = (runId, name) => wdb.getRunState(runId).nodes.find((n) => n.name === name);

  it('executor 抛异常：节点 fail，且 timeline 里查得到原因（不是空的）', async () => {
    const runId = buildRun(['prd-planning']);
    engine.setAgentExecutor(async () => {
      // 真跑第一个 work 节点时 AgentManager.getAgentConfig 抛的就是这一句
      throw new Error('Agent not found: prd-planning');
    });

    engine.startRun(runId);
    await engine.nextStep(runId);

    const node = nodeByName(runId, 'prd-planning');
    // 没有 fail 边的死路：fail 之后就地 blocked —— 真跑第一个 work 节点时看到的就是这个
    expect(node.status).toBe('blocked');
    // 引擎仍然不发明 fail_reason（它是选边的匹配键）
    expect(node.failReason).toBeNull();

    const errs = timeline(runId).filter((e) => e.kind === 'node.error');
    expect(errs).toHaveLength(1);
    expect(errs[0].summary).toContain('Agent not found: prd-planning');
    expect(errs[0].payload.node_run_id).toBe(node.id);
    expect(errs[0].payload.attempt).toBe(0);
    expect(errs[0].payload.node_name).toBe('prd-planning');
    expect(errs[0].payload.system_agent_id).toBe('prd-planning');
    expect(errs[0].payload.error).toBe('Agent not found: prd-planning');
    expect(errs[0].payload.error_type).toBe('Error');
    expect(errs[0].payload.error_stack).toContain('Error: Agent not found');
    // ref = 节点 run id：前端 NodeRunLog 按 node_run_id 归组，这条落在失败节点自己的卡片里
    expect(errs[0].ref).toBe(node.id);

    // 同一条事件也在实时轨上（emitEvent 保证两轨不分家）
    const onStream = streamEvents(runId).filter((e) => e.event === 'node.error');
    expect(onStream).toHaveLength(1);
    expect(onStream[0].data.error).toBe('Agent not found: prd-planning');
  });

  it('抛异常时 node.output 事件带 output_excerpt，错误原因同样能读回', async () => {
    const runId = buildRun(['a']);
    engine.setAgentExecutor(async () => { throw new Error('boom: 配置缺失'); });

    engine.startRun(runId);
    await engine.nextStep(runId);

    const out = streamEvents(runId).filter((e) => e.event === 'node.output');
    expect(out).toHaveLength(1);
    expect(out[0].data.status).toBe('fail');
    expect(out[0].data.fail_reason).toBeNull();
    expect(out[0].data.output_excerpt).toBe('boom: 配置缺失');
  });

  it('executor 正常返回的 output 也进 output_excerpt，超长按定长截断', async () => {
    const runId = buildRun(['long']);
    const huge = 'x'.repeat(5000);
    engine.setAgentExecutor(async () => ({ status: 'success', output: huge }));

    engine.startRun(runId);
    await engine.nextStep(runId);

    const out = streamEvents(runId).find((e) => e.event === 'node.output');
    expect(out.data.status).toBe('success');
    expect(out.data.output_excerpt.startsWith('xxxx')).toBe(true);
    // 截断了，且短得多（完整输出只在 output_ref 指的产物目录里，不进事件也不进库）
    expect(out.data.output_excerpt.length).toBeLessThan(huge.length);
    expect(out.data.output_excerpt).toContain('已截断');
  });

  it('executor 没有 output 时 output_excerpt 是 null，不塞空串', async () => {
    const runId = buildRun(['quiet']);
    engine.setAgentExecutor(async () => ({ status: 'success' }));

    engine.startRun(runId);
    await engine.nextStep(runId);

    const out = streamEvents(runId).find((e) => e.event === 'node.output');
    expect(out.data.output_excerpt).toBeNull();
    // 没抛异常就不该有 node.error
    expect(timeline(runId).filter((e) => e.kind === 'node.error')).toHaveLength(0);
  });
});