/** * 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); }); });