宜博 / lemonboss

/**
 * Unit Tests — 恢复动作在 **HTTP 这一头**的结局（27-P3C 批 3 对抗复核 F1 / 409）
 *
 * 引擎侧的两种结局已经有用例钉着（graph-run-slot-recovery.test.js），但用户点的是按钮、
 * 外部 agent 打的是 curl —— 真正决定「这次恢复有没有可执行的下一步」的是**响应**：
 *   · 退得回队列的：必须是 2xx + 一个说得清「它回队列了」的体。回 500 的话前端只会当故障，
 *     而库里其实已经排好队了（状态和响应对不上，正是最难查的一类）；
 *   · 退不回队列的：必须是 409 + 点名占坑者。500 的含义是「服务端坏了」，调用方会去重试
 *     或报故障，而这里唯一有用的动作是「等它收敛 / 去停掉它」。
 *
 * 为什么非得走真 express + fetch：`sendError` 是私有函数，且 `/start` 是**同步** handler ——
 * `h()` 早先的 `Promise.resolve(fn()).catch()` 挂 catch 时 fn 已经抛完了，异常直接冒到 express
 * 默认处理器、回一页 HTML 的 500，前端 `await r.json()` 炸在 "Unexpected token '<'"。
 * 只有从 HTTP 这一头才测得出这件事（已有的 409 用例走的是 `/nodes/:id/retry` 那条，
 * 这里换 `/start`：四个同步端点各自的 try/catch 是分别写的，一条路通不代表另一条通）。
 *
 * 每个用例一家自己的公司 —— 「坑」是 per-company 的，各测各的互不干扰，也就不存在用例顺序依赖。
 */
const fs = require('fs');
const os = require('os');
const path = require('path');
const express = require('express');

const TEST_DB_PATH = '/tmp/test-p3c-recovery-http.sqlite';
const DB_FILES = [TEST_DB_PATH, `${TEST_DB_PATH}-shm`, `${TEST_DB_PATH}-wal`];
const MODULES = [
  '../../services/graph-db',
  '../../services/org-graph-service',
  '../../services/work-graph-db',
  '../../services/company-paths',
  '../../services/graph-run-sse',
  '../../services/graph-engine',
  '../../services/task-settle',
  '../../services/task-scheduler',
  '../../routes/graph-run',
];

describe('恢复动作的 HTTP 结局：退得回队列 → 2xx，退不回 → 409（都不是 500）', () => {
  let wdb;
  let engine;
  let scheduler;
  let server;
  let origin;
  let runsDir;
  /** 挂起的 executor promise，afterAll 统一放掉 */
  const pendingResolvers = [];

  beforeAll(async () => {
    for (const f of DB_FILES) { try { fs.unlinkSync(f); } catch { /* 本来就没有 */ } }
    process.env.GRAPH_DB_PATH = TEST_DB_PATH;
    runsDir = fs.mkdtempSync(path.join(os.tmpdir(), 'p3c-http-runs-'));
    process.env.GRAPH_RUNS_DIR = runsDir;
    for (const m of MODULES) delete require.cache[require.resolve(m)];
    wdb = require('../../services/work-graph-db');
    engine = require('../../services/graph-engine');
    scheduler = require('../../services/task-scheduler');

    const app = express();
    app.use(express.json());
    app.use('/api/graph-runs', require('../../routes/graph-run'));
    await new Promise((resolve) => { server = app.listen(0, resolve); });
    origin = `http://127.0.0.1:${server.address().port}`;
  });

  afterAll(async () => {
    for (const r of pendingResolvers) r();
    await new Promise((r) => setTimeout(r, 100));
    if (server) server.close();
    scheduler.stopSweeper();
    cleanupLeakedCompanyDirs();
    for (const f of DB_FILES) { try { fs.unlinkSync(f); } catch { /* 已经删了 */ } }
    try { fs.rmSync(runsDir, { recursive: true, force: true }); } catch { /* 同上 */ }
    delete process.env.GRAPH_DB_PATH;
    delete process.env.GRAPH_RUNS_DIR;
    for (const m of MODULES) delete require.cache[require.resolve(m)];
  });

  /** 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('p3chttp-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 { /* 删不掉就留着 */ }
    }
  }

  let seq = 0;
  const nextCompany = () => `p3chttp-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, 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 buildGraph(companyId, node) {
    ensureAgents([node]);
    const g = wdb.createWorkGraph({ companyId, nodes: [node], edges: [] });
    expect(g.errors || []).toEqual([]);
    return g.id;
  }

  /** 占坑的那条：一条**有 task** 的 running root run（409 的 holders 要带得出 taskId） */
  function occupySlot(companyId, tag) {
    const goalId = wdb.createGoal({ companyId, title: 'g', description: 'g' }).id;
    const taskId = wdb.createTask({ goalId, companyId, title: `holder-${tag}`, description: '占坑的任务' }).id;
    const run = wdb.createRun({
      graphKind: 'work', graphId: buildGraph(companyId, `h${tag}`), companyId, goalId, taskId,
    });
    wdb.getDb().prepare(`UPDATE graph_runs SET run_status = 'running' WHERE id = ?`).run(run.id);
    wdb.updateTask(taskId, { status: 'running', currentRunId: run.id });
    return { runId: run.id, taskId, goalId };
  }

  /** 预算耗尽那种局面：一条 blocked 的 root run + 一条 blocked 的 task（人正要点「启动」） */
  function blockedTaskRun(companyId, goalId, tag) {
    const taskId = wdb.createTask({
      goalId, companyId, title: `t-${tag}`, description: '预算耗尽卡住的任务', executionMode: 'auto',
    }).id;
    const run = wdb.createRun({
      graphKind: 'work', graphId: buildGraph(companyId, `n${tag}`), companyId, goalId, taskId,
      executionMode: 'auto',
    });
    wdb.getDb().prepare(`UPDATE graph_runs SET run_status = 'blocked' WHERE id = ?`).run(run.id);
    wdb.updateTask(taskId, {
      status: 'blocked', currentRunId: run.id, blockedReason: 'budget_exhausted',
      queuedAt: new Date().toISOString(),
    });
    return { runId: run.id, taskId };
  }

  /** 永不 resolve 的 executor：被接回来的那条稳稳停在 running，断言不跟异步驱动赛跑 */
  function hangingExecutor() {
    engine.setAgentExecutor(() => new Promise((resolve) => {
      pendingResolvers.push(() => resolve({ status: 'success' }));
    }));
  }

  /** 等一个条件成立（调度是 setImmediate + 异步驱动，不能只 flush 一次就断言） */
  async function waitFor(fn, ms = 1000) {
    const deadline = Date.now() + ms;
    for (;;) {
      const v = fn();
      if (v) return v;
      if (Date.now() > deadline) return v;
      await new Promise((r) => setTimeout(r, 10));
    }
  }

  it('退不回队列的恢复 → 409 + 点名占坑者，绝不是 500（/start 是同步 handler，早先直接冒到 express）', async () => {
    const companyId = nextCompany();
    const holder = occupySlot(companyId, 'a');
    // 没有 task 的裸 root run：队列按 task 选候选，它退不回去 —— 这时唯一正确的答复是 409
    const bare = wdb.createRun({ graphKind: 'work', graphId: buildGraph(companyId, 'bare'), companyId });
    wdb.getDb().prepare(`UPDATE graph_runs SET run_status = 'blocked' WHERE id = ?`).run(bare.id);

    const r = await fetch(`${origin}/api/graph-runs/${bare.id}/start`, { method: 'POST' });
    expect(r.headers.get('content-type')).toMatch(/application\/json/); // 500 那条路回的是一页 HTML
    const body = await r.json();

    expect(r.status).toBe(409);
    expect(body.reason).toBe('run-slot-busy');
    expect(body.companyId).toBe(companyId);
    expect(body.holders).toEqual([
      expect.objectContaining({ runId: holder.runId, taskId: holder.taskId, runStatus: 'running' }),
    ]);
    // hint 必须指向此刻真的走得通的一步：占坑那条是 running，/stop 认它
    expect(body.hint).toContain(`/api/graph-runs/${holder.runId}/stop`);
    // 拒绝是干净的：一个字都没写进去
    expect(wdb.getRun(bare.id).run_status).toBe('blocked');
  });

  it('人对卡住的任务点「启动」撞上坑满 → 2xx + 明说它回了队列（不是 500，也不是原地卡死）', async () => {
    const companyId = nextCompany();
    const goalId = wdb.createGoal({ companyId, title: 'g', description: 'g' }).id;
    occupySlot(companyId, 'b');
    const t1 = blockedTaskRun(companyId, goalId, 'b');

    const r = await fetch(`${origin}/api/graph-runs/${t1.runId}/start`, {
      method: 'POST',
      headers: { 'content-type': 'application/json' },
      body: JSON.stringify({ mode: 'auto' }),
    });
    const body = await r.json();

    expect(r.status).toBe(200);
    expect(body).toMatchObject({ ok: true, requeued: true, runStatus: 'queued' });
    // 响应说什么，库里就得是什么 —— run 与 task 两行一起回队列，少一半都是新的卡死形态
    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();
  });

  it('坑一空，刚才那条自己被接回来 —— 人只点了一次「启动」，没有第二次操作', async () => {
    const companyId = nextCompany();
    const goalId = wdb.createGoal({ companyId, title: 'g', description: 'g' }).id;
    const holder = occupySlot(companyId, 'c');
    const t1 = blockedTaskRun(companyId, goalId, 'c');
    hangingExecutor();

    expect((await fetch(`${origin}/api/graph-runs/${t1.runId}/start`, { method: 'POST' })).status).toBe(200);
    expect(wdb.getRun(t1.runId).run_status).toBe('queued');

    // 人去把占坑那条停掉（正是 409 的 hint 指的那条路），之后**不再点任何按钮**
    const stopped = await fetch(`${origin}/api/graph-runs/${holder.runId}/stop`, { method: 'POST' });
    expect(stopped.status).toBe(200);

    await waitFor(() => wdb.getRun(t1.runId).run_status === 'running');
    expect(wdb.getRun(t1.runId).run_status).toBe('running');
    expect(wdb.getTask(t1.taskId).status).toBe('running');
  });
});