宜博 / lemonboss

/**
 * 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(); // 审计轨迹还在
  });
});