宜博 / lemonboss

/**
 * Unit Tests — task 转 blocked 时主动唤醒 ceo（39 号文档 §二 P3，事件驱动主路径）
 *
 * 背景：blocked 不是终态，`settleGoalIfAllTasksDone` 不会碰它，goal 因此不会收敛，
 * `reportGoalOutcome` 那条唤醒 report_to 的路径也不会触发——卡住的任务本来会一直静默
 * 躺着，直到有人自己点进控制台才发现（30 号文档点过名的空白）。`settleRootRun` 现在
 * 会在 blocked 回写之后顺手唤醒本公司的 ceo。
 *
 * 只验唤醒这一跳的逻辑（URL / header / body 形状），不验 agent-chat 真的收到消息之后
 * 会做什么——那是 ceo 自己 tools.md / heartbeat.md 的行为，不在引擎测试射程内。
 *
 * 隔离手法与 task-blocked-reason.test.js 一致。
 */
import { describe, it, expect, beforeAll, afterAll, beforeEach, vi } from 'vitest';
import fs from 'fs';

describe('task 转 blocked 唤醒 ceo（39 号文档 P3）', () => {
  let wdb;
  let settle;
  const TEST_DB_PATH = '/tmp/test-blocked-ceo-wake.sqlite';
  const files = [TEST_DB_PATH, `${TEST_DB_PATH}-shm`, `${TEST_DB_PATH}-wal`];

  beforeAll(() => {
    for (const f of files) { try { fs.unlinkSync(f); } catch { /* 没有就算了 */ } }
    process.env.GRAPH_DB_PATH = TEST_DB_PATH;
    for (const m of [
      '../../services/graph-db',
      '../../services/org-graph-service',
      '../../services/work-graph-db',
      '../../services/graph-run-sse',
      '../../services/task-settle',
    ]) {
      delete require.cache[require.resolve(m)];
    }
    wdb = require('../../services/work-graph-db');
    settle = require('../../services/task-settle');
  });

  afterAll(() => {
    for (const f of files) { try { fs.unlinkSync(f); } catch { /* 同上 */ } }
    delete process.env.GRAPH_DB_PATH;
  });

  beforeEach(() => {
    const db = wdb.getDb();
    db.pragma('foreign_keys = OFF');
    db.exec(`
      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');
  });

  let seq = 0;

  /** 与 task-blocked-reason.test.js 的 scene() 同构：造出「一条 root run 正卡在某节点」 */
  function scene() {
    const companyId = `ceowake-co-${++seq}`;
    const now = new Date().toISOString();
    const db = wdb.getDb();
    db.prepare(`
      INSERT OR IGNORE INTO system_agents
        (id, name, scope, owner_company_id, legacy_path, status, created_at, updated_at)
      VALUES ('sdd-planning', 'sdd-planning', 'public', NULL, 'system/agents/sdd-planning', 'active', ?, ?)
    `).run(now, now);

    const graph = wdb.createWorkGraph({ companyId, nodes: ['sdd-planning'], edges: [] });
    const goal = wdb.createGoal({ companyId, title: '做一个官网', description: '测试用' });
    const task = wdb.createTask({
      goalId: goal.id, companyId, title: '做一个官网', description: '测试用', executionMode: 'manual',
    });
    const run = wdb.createRun({
      graphKind: 'work', graphId: graph.id, companyId, goalId: goal.id, taskId: task.id,
      executionMode: 'manual', runStatus: 'running',
    });
    wdb.syncRunNodesToDef(run.id);
    wdb.updateTask(task.id, { status: 'running', currentRunId: run.id, startedAt: now });

    const nodeRow = db.prepare(`SELECT id FROM graph_run_nodes WHERE run_id = ?`).get(run.id);
    db.prepare(`UPDATE graph_run_nodes SET status = 'blocked', fail_reason = 'boundary_escape', updated_at = ? WHERE id = ?`)
      .run(now, nodeRow.id);

    return { companyId, taskId: task.id, run: wdb.getRun(run.id) };
  }

  it('blocked 回写时会 fetch 唤醒 `{companyId}:ceo`，带上正确的内部令牌头', async () => {
    const { companyId, run } = scene();
    const { INTERNAL_TOKEN, INTERNAL_TOKEN_HEADER } = require('../../middleware/jwt-auth');

    const calls = [];
    const originalFetch = global.fetch;
    global.fetch = vi.fn((url, opts) => {
      calls.push({ url, opts });
      return Promise.resolve({ ok: true, status: 200 });
    });

    try {
      settle.settleRootRun(run, 'blocked');
      // wakeCeo 是发射后不等结果，让微任务队列跑一轮再断言
      await new Promise((r) => setImmediate(r));
    } finally {
      global.fetch = originalFetch;
    }

    expect(calls).toHaveLength(1);
    const [{ url, opts }] = calls;
    expect(url).toContain(`/api/agent/${companyId}%3Aceo/chat`);
    expect(opts.method).toBe('POST');
    expect(opts.headers[INTERNAL_TOKEN_HEADER]).toBe(INTERNAL_TOKEN);
    const body = JSON.parse(opts.body);
    expect(body.message).toContain('blocked');
    expect(body.isHeartbeat).toBe(true);
  });

  it('completed 回写不会唤醒 ceo（只有 blocked 才需要主动叫人）', async () => {
    const { run } = scene();
    const calls = [];
    const originalFetch = global.fetch;
    global.fetch = vi.fn((url, opts) => { calls.push({ url, opts }); return Promise.resolve({ ok: true }); });

    try {
      // 先把节点状态改成 success，让 settleTaskFromRun 认成 completed
      const db = wdb.getDb();
      db.prepare(`UPDATE graph_run_nodes SET status = 'success', fail_reason = NULL WHERE run_id = ?`).run(run.id);
      settle.settleRootRun(run, 'completed');
      await new Promise((r) => setImmediate(r));
    } finally {
      global.fetch = originalFetch;
    }

    expect(calls).toHaveLength(0);
  });

  it('fetch 唤醒失败（网络异常）不影响 settleRootRun 的返回值 —— best-effort，不能拖垮主路径', () => {
    const { run } = scene();
    const originalFetch = global.fetch;
    global.fetch = vi.fn(() => Promise.reject(new Error('网络挂了')));

    let result;
    try {
      result = settle.settleRootRun(run, 'blocked');
    } finally {
      global.fetch = originalFetch;
    }

    expect(result).toEqual({ task: 'blocked', goal: null });
  });

  it('低频兜底扫描 sweepBlockedTasksForCeo：按公司分组只唤醒一次，不逐条 task 叫', async () => {
    const a = scene();
    const b = scene();
    // 同一家公司再卡一条 —— a.companyId 应该只被唤醒一次，不是两次
    const db = wdb.getDb();
    const now = new Date().toISOString();
    const graph2 = wdb.createWorkGraph({ companyId: a.companyId, nodes: ['sdd-planning'], edges: [] });
    const goal2 = wdb.createGoal({ companyId: a.companyId, title: '第二条', description: '测试用' });
    const task2 = wdb.createTask({
      goalId: goal2.id, companyId: a.companyId, title: '第二条', description: '测试用', executionMode: 'manual',
    });
    wdb.updateTask(task2.id, { status: 'blocked' });
    // scene() 已经把 a/b 的 task 结算成 blocked（settleRootRun 还没跑，这里直接跑一次让状态落到位）
    settle.settleRootRun(a.run, 'blocked');
    settle.settleRootRun(b.run, 'blocked');

    const calls = [];
    const originalFetch = global.fetch;
    global.fetch = vi.fn((url) => { calls.push(url); return Promise.resolve({ ok: true }); });

    try {
      const rows = settle.sweepBlockedTasksForCeo();
      await new Promise((r) => setImmediate(r));
      const companiesSwept = rows.map((r) => r.company_id).sort();
      expect(companiesSwept).toEqual([a.companyId, b.companyId].sort());
      const aRow = rows.find((r) => r.company_id === a.companyId);
      expect(aRow.cnt).toBe(2); // a 公司这次有 2 条 blocked task，但下面只应该 fetch 一次
    } finally {
      global.fetch = originalFetch;
    }

    const aCalls = calls.filter((u) => u.includes(`${a.companyId}%3Aceo`));
    expect(aCalls).toHaveLength(1);
  });
});