/** * Unit Tests — task-scheduler.js(27-P3C 批 3 · 排队) * * 逐条对应 §十 验收: * 2 T1 收敛后 T2 自动起跑(漏斗钩子覆盖正常收敛) * 13 停止即出队(stopRun 不走 settleRun,钩子必须覆盖它) * 15 排队期间 replan → 出队后按新图跑(syncRunNodesToDef) * 23 上游失败 → 下游 blocked,上游恢复后自动回 queued(blocked 是推导出来的,不落死) * 外加 §九.1 那张「queued 混进 run_status」的回归表里点名要钉死的三条断言: * · queued 不在活跃集里(排队不占坑,出队那一跳才被准入事务数一次) * · advanceOne 挡住 queued run(防止 queued run 被误推进的唯一一道闸) * · startRun 拒绝 queued run(出队顺序不能反过来) * * 隔离手法与 graph-engine.test.js 一致:先设 GRAPH_DB_PATH / GRAPH_RUNS_DIR 再 require * (graph-run-sse 在模块加载时快照 GRAPH_RUNS_DIR,先 require 再设会往仓库 data/companies/ 写, * 而且不报错)。agent 一律走注入的假 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('task-scheduler', () => { let wdb; let engine; let sse; let scheduler; let runsDir; const TEST_DB_PATH = '/tmp/test-task-scheduler.sqlite'; /** 挂起的 executor promise,afterAll 统一放掉,免得留下永不 settle 的 driver */ 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(), 'task-scheduler-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/ 下给每家公司建项目根 * (它不认 GRAPH_RUNS_DIR)。只删本文件造的那些前缀目录,且必须确认里面没有 company.json * —— 那是真公司,删了就是删用户数据。(写法与 goal-task-acceptance.test.js 一致) */ function cleanupLeakedCompanyDirs() { const root = path.resolve(__dirname, '../../../data/companies'); if (!fs.existsSync(root)) return; for (const cid of fs.readdirSync(root)) { if (!cid.startsWith('sched-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 () => { // 先放掉挂起的 executor 再等一拍:auto driver 恢复后还会跑一轮 advanceOne, // 而 ensureProjectRoot 会把刚删掉的公司目录原样建回来 —— 顺序反了就等于没清理 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; /** 每个用例一家独立公司:执行坑是按 company 数的 */ const nextCompany = () => `sched-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); } /** 建一张链式 work graph 定义(校验要求单入口 + 全可达,所以多节点时串起来) */ 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; } /** * 「入队」——批 3 阶段一没有端点,这里手工拼出端点将来要写的那三步: * 建 task 行 → 建 run 行(**runStatus: 'queued'**,不占坑)→ task 标 queued + 挂 current_run_id。 */ function enqueue(companyId, goalId, { title, mode = 'auto', dependsOn = [], priority = 'P1', graphId } = {}) { const task = wdb.createTask({ goalId, companyId, title, description: `${title} 的描述`, executionMode: mode, dependsOn, priority, }); const run = wdb.createRun({ graphKind: 'work', graphId: graphId || buildGraph(companyId), companyId, goalId, taskId: task.id, executionMode: mode, 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() { const calls = []; engine.setAgentExecutor((ctx) => { calls.push(ctx.nodeName); return new Promise((resolve) => pending.push(() => resolve({ status: 'success' }))); }); return calls; } const flush = () => new Promise((r) => setImmediate(() => setImmediate(r))); // ────────────────────────────────────────────────────────────── // §九.1 回归表:queued 不被误当可执行 // ────────────────────────────────────────────────────────────── describe('queued 混进 run_status 的三道闸', () => { it('queued 不占坑:同一公司可以躺任意多条 queued root run,而出队时超发被准入事务拒', () => { const companyId = nextCompany(); const goalId = newGoal(companyId); // 这条断言是出队原子性的**全部依赖**,钉死它 expect(wdb.ACTIVE_ROOT_RUN_STATUS).toEqual(['ready', 'running', 'paused']); expect(wdb.ACTIVE_ROOT_RUN_STATUS).not.toContain('queued'); const a = enqueue(companyId, goalId, { title: 'a' }); const b = enqueue(companyId, goalId, { title: 'b' }); const c = enqueue(companyId, goalId, { title: 'c' }); for (const r of [a, b, c]) expect(wdb.getRun(r.runId).run_status).toBe('queued'); // 第一条出队 = CAS 成功 expect(wdb.updateRunStatus(a.runId, 'queued', 'ready')).toBe(1); // 第二条被**主守卫**(准入事务里数坑)拒:抛 RunSlotBusyError,**不是** changes = 0 expect(() => wdb.updateRunStatus(b.runId, 'queued', 'ready')).toThrow(wdb.RunSlotBusyError); expect(wdb.getRun(b.runId).run_status).toBe('queued'); // 批 3 这里演示的是「兜底那一层(唯一索引)也还在」:绕开 updateRunStatus 直接写库照样被拒。 // 批 4 DROP 了那条索引,所以这一段反过来钉两件事 —— // ① 索引确实没了:裸写库现在**写得进去**(这正是「删掉了最后一道 DB 兜底」的样子, // 要能演示出来,否则「已 DROP」只是一句注释); // ② 于是唯一的守卫只剩上面那个准入事务,它今天就得是对的。 expect(() => wdb.getDb().prepare(`UPDATE graph_runs SET run_status = 'ready' WHERE id = ?`).run(b.runId) ).not.toThrow(); expect(wdb.getRun(b.runId).run_status).toBe('ready'); // 把裸写的这条放回 queued,别让它污染后面的断言(它是绕过守卫写进去的,不是准入放行的) wdb.getDb().prepare(`UPDATE graph_runs SET run_status = 'queued' WHERE id = ?`).run(b.runId); // 同一条重复出队 = changes 0(幂等 no-op,不抛) expect(wdb.updateRunStatus(a.runId, 'queued', 'ready')).toBe(0); // queued 有出路:撤出队列 expect(wdb.canTransitionRun('queued', 'stopped')).toBe(true); expect(wdb.updateRunStatus(c.runId, 'queued', 'stopped')).toBe(1); }); /** * 准入守卫(updateRunStatus 里的 BEGIN IMMEDIATE 事务)。 * 判据是「**从活跃集外进入活跃集**」—— 这几条把「按 to 判」那种写法钉死在墙上: * 按 to 判的话 ready→running 会把 run 自己算成竞争者,第一条 run 都起不来。 */ describe('准入守卫:从活跃集外进入活跃集才占坑', () => { it('集内移动不占新坑:ready→running / running→paused / paused→running 一路畅通', () => { const companyId = nextCompany(); const { runId } = enqueue(companyId, newGoal(companyId), { title: 'a' }); expect(wdb.updateRunStatus(runId, 'queued', 'ready')).toBe(1); expect(wdb.countActiveRootRuns(companyId)).toBe(1); // 坑已经被它自己占着 // 下面三跳的 from 全在活跃集里 —— 计数命中的是它自己,按 to 判就会在这里全军覆没 expect(wdb.updateRunStatus(runId, 'ready', 'running')).toBe(1); expect(wdb.updateRunStatus(runId, 'running', 'paused')).toBe(1); expect(wdb.updateRunStatus(runId, 'paused', 'running')).toBe(1); expect(wdb.countActiveRootRuns(companyId)).toBe(1); // 全程一个坑 }); it('活跃集外进来的才拦:blocked→running 在坑满时被拒,坑空了同一跳就过', () => { const companyId = nextCompany(); const goalId = newGoal(companyId); const a = enqueue(companyId, goalId, { title: 'a' }); const b = enqueue(companyId, goalId, { title: 'b' }); wdb.updateRunStatus(a.runId, 'queued', 'ready'); wdb.updateRunStatus(a.runId, 'ready', 'running'); // b 直接落到 blocked(活跃集外,不占坑;queued→blocked 不是合法转换,所以裸写库造状态) wdb.getDb().prepare(`UPDATE graph_runs SET run_status = 'blocked' WHERE id = ?`).run(b.runId); expect(wdb.countActiveRootRuns(companyId)).toBe(1); expect(() => wdb.updateRunStatus(b.runId, 'blocked', 'running')).toThrow(wdb.RunSlotBusyError); // a 让出坑(running → completed 离开活跃集) expect(wdb.updateRunStatus(a.runId, 'running', 'completed')).toBe(1); expect(wdb.updateRunStatus(b.runId, 'blocked', 'running')).toBe(1); }); it('子 run 不占公司的坑:root 正在跑时,子 run 照样能 planning→ready', () => { const companyId = nextCompany(); const { runId } = enqueue(companyId, newGoal(companyId), { title: 'a' }); wdb.updateRunStatus(runId, 'queued', 'ready'); const child = wdb.createRun({ graphKind: 'work', graphId: buildGraph(companyId, ['child']), companyId, parentRunId: runId, runStatus: 'planning', }); expect(wdb.updateRunStatus(child.id, 'planning', 'ready')).toBe(1); // 计数只认 root:坑还是那一个 expect(wdb.countActiveRootRuns(companyId)).toBe(1); }); /** * 这一条以前钉的是「坑满的报错文本仍能被 company.js 的 `/UNIQUE|constraint/i` 嗅到」, * 理由写着「routes/company.js 那两处 catch 至今是按文本嗅探的」—— **批 3 把那两处 * catch 整段删了**(下达不再因坑满失败,改成入队)。于是这条用例一直绿着,钉的却是 * 一个没有任何调用方的契约,还反过来禁止清理那句报文。 * 现在钉真正的那条:判据只有 `work-graph-db.isSlotRaceError` 一个持有者,按类型判。 * 该认的一个都不能漏 —— 漏一种,调度器就会把一次正常竞争当故障去 abandonRun(丢任务), * 而路由那边照常 409,两边对同一件事给出相反结论。 * 不该认的也一个都不能多认(批 4 起 SQLITE_CONSTRAINT_UNIQUE 就在这一侧,见下): * 认宽了是把真 bug 当正常竞争 continue 掉,那是静默。 */ it('「没抢到坑」的判据只有一个持有者,且按类型判 —— 改一句报错文案不该改变任何分支', () => { const companyId = nextCompany(); const goalId = newGoal(companyId); const a = enqueue(companyId, goalId, { title: 'a' }); const b = enqueue(companyId, goalId, { title: 'b' }); wdb.updateRunStatus(a.runId, 'queued', 'ready'); let caught; try { wdb.updateRunStatus(b.runId, 'queued', 'ready'); } catch (e) { caught = e; } expect(caught).toBeInstanceOf(wdb.RunSlotBusyError); expect(wdb.isSlotRaceError(caught)).toBe(true); // 批 3 这里钉的是「兜底那一层(唯一索引)抛的 SqliteError 也必须被认成『没抢到』」。 // 批 4 DROP 了索引,于是这条判据**反过来**:一个 SQLITE_CONSTRAINT_UNIQUE 再出现在 // 这条路上,只可能是索引被旧版本进程建回来了(max>1 被悄悄压回 1)。 // 认成「没抢到」的话调度器会 continue 掉,任务永远排队、日志一行没有 —— // 正是要消灭的「写不进去但不报错」。所以它必须**不**算竞争,一路抛到人能看见的地方。 const fakeUniqueErr = Object.assign(new Error('UNIQUE constraint failed: graph_runs.company_id'), { code: 'SQLITE_CONSTRAINT_UNIQUE', }); expect(wdb.isSlotRaceError(fakeUniqueErr)).toBe(false); // ⚠️ 批 3 这里断言 InvalidTransitionError 也算「没抢到」,注释写着「这条 run 已经被别人 // 从 queued 推走了」。**探针实测证伪**:被别人推走时 updateRunStatus 走的是条件更新 // `WHERE id=? AND run_status=@from`,**返回 0、不抛异常**;真正抛 InvalidTransitionError // 的只有「调用方声明了非法转换」(expected 是调用方传进来的声明值,canTransitionRun // judge 的是 expected→to 而不是 cur.run_status→to)。 // 把调用方的 bug 当成正常竞争吞掉、还不留日志 —— 与上面那条 UNIQUE 同一个病。 expect(wdb.isSlotRaceError(new wdb.InvalidTransitionError('run', 'x', 'queued', 'ready'))).toBe(false); // 别的错误一律不算「没抢到」—— 认宽了就是把真 bug 当成正常竞争 continue 掉 expect(wdb.isSlotRaceError(new Error('UNIQUE constraint failed: 别处的错'))).toBe(false); expect(wdb.isSlotRaceError(null)).toBe(false); // 两个调用方都必须走这一份,不许再各写一份(已经分叉过一次了)。 // 读源码而不是调它们的私有函数:这两个 isRaceError / isSlotRace 都没导出, // 而「有没有第二个持有者」本来就是源码层面的事实。 for (const mod of ['../../services/task-scheduler.js', '../../routes/company.js']) { const src = fs.readFileSync(path.resolve(__dirname, mod), 'utf8'); expect(src).toContain('isSlotRaceError'); // 报错文本嗅探不许回来:它正是上一版分叉的那一半 expect(src).not.toMatch(/\/UNIQUE constraint failed\/i\.test/); } }); it('「别人抢先做完了」不许报成「坑满」:CAS 没匹配上返回 0,不抛', () => { const companyId = nextCompany(); const { runId } = enqueue(companyId, newGoal(companyId), { title: 'a' }); expect(wdb.updateRunStatus(runId, 'queued', 'ready')).toBe(1); // 再来一次同样的出队:此刻 countActiveRootRuns 已经是 1,少了事务内那次重读 // 就会把这次幂等 no-op 误报成 RunSlotBusyError —— 调度器据此会去 abandonRun, // 把一条刚刚成功起跑的任务收成 cancelled。 expect(wdb.updateRunStatus(runId, 'queued', 'ready')).toBe(0); }); }); /** * 批 4:额度从「模块级常量 1」变成 per-company 的 `company.json.max_parallel_tasks`。 * * 这组用例的重点不是「并行能跑」,而是**回落**那一条:全仓几乎所有单测的公司都没有 * company.json(`sched-co-*` 只是个字符串),一旦回落算错,1195 条测试会一起变色 —— * 所以「读不到 = 1」是保护存量的那条断言,必须先钉。 * * 配置写在 `company-paths.companyRootDir()` 下而不是 `data/companies/` 下: * 前者认 GRAPH_RUNS_DIR,本文件设了它,于是这组用例不会去碰仓库里的真公司目录 * (准入判据一旦去读真 data/,撞上同名公司就读到真配置 —— §六 缺陷 A 同型)。 */ describe('每家公司自己的执行坑数(company.json.max_parallel_tasks)', () => { /** 给这家公司写一份 company.json;`n === null` 表示写一个没有该字段的配置 */ function setMaxParallel(companyId, n) { const { companyRootDir } = require('../../services/company-paths'); const dir = companyRootDir(companyId); fs.mkdirSync(dir, { recursive: true }); const cfg = { id: companyId, name: companyId }; if (n !== null) cfg.max_parallel_tasks = n; fs.writeFileSync(path.join(dir, 'company.json'), JSON.stringify(cfg)); return companyId; } it('没有 company.json 的公司回落 1 —— 存量单测与存量公司的行为一个字不变', () => { const companyId = nextCompany(); expect(wdb.maxActiveRootRuns(companyId)).toBe(wdb.DEFAULT_MAX_ACTIVE_ROOT_RUNS); expect(wdb.DEFAULT_MAX_ACTIVE_ROOT_RUNS).toBe(1); // 有 company.json 但没写这个字段,同样回落 expect(wdb.maxActiveRootRuns(setMaxParallel(nextCompany(), null))).toBe(1); }); it('非法值一律回落 1,不静默当成无限:0 / 负数 / 小数 / 字符串 / 超上限', () => { for (const bad of [0, -3, 1.5, '3', true, null]) { expect(wdb.maxActiveRootRuns(setMaxParallel(nextCompany(), bad))).toBe(1); } // 超上限按上限截断(不是回落 1,也不是当真) const huge = setMaxParallel(nextCompany(), 100000); expect(wdb.maxActiveRootRuns(huge)).toBe(wdb.MAX_ACTIVE_ROOT_RUNS_CAP); }); it('max=3:前三条准入放行,第 4 条被拒且 active / max / heldBy 三个字段都对', () => { const companyId = setMaxParallel(nextCompany(), 3); const goalId = newGoal(companyId); const runs = ['a', 'b', 'c', 'd'].map((title) => enqueue(companyId, goalId, { title }).runId); for (const [i, runId] of runs.slice(0, 3).entries()) { expect(wdb.updateRunStatus(runId, 'queued', 'ready')).toBe(1); expect(wdb.countActiveRootRuns(companyId)).toBe(i + 1); } let caught; try { wdb.updateRunStatus(runs[3], 'queued', 'ready'); } catch (e) { caught = e; } expect(caught).toBeInstanceOf(wdb.RunSlotBusyError); expect(caught.max).toBe(3); expect(caught.active).toBe(3); // heldBy 点得出**具体是谁**占着 —— 少了它 409 上只剩「坑满」,人无从判断该等谁 expect(caught.heldBy.slice().sort()).toEqual(runs.slice(0, 3).slice().sort()); expect(wdb.getRun(runs[3]).run_status).toBe('queued'); // 拒绝是干净的 // 让出一个坑,第 4 条同一跳立刻过(额度是「同时几条」,不是「一共几条」) expect(wdb.updateRunStatus(runs[0], 'ready', 'running')).toBe(1); // 集内移动,不占新坑 expect(wdb.updateRunStatus(runs[0], 'running', 'completed')).toBe(1); expect(wdb.updateRunStatus(runs[3], 'queued', 'ready')).toBe(1); }); it('额度是 per-company 的:A 公司放开到 3,B 公司照样只有 1', () => { const a = setMaxParallel(nextCompany(), 3); const b = nextCompany(); // 没配 → 1 const aRuns = ['a1', 'a2'].map((t) => enqueue(a, newGoal(a), { title: t }).runId); const bRuns = ['b1', 'b2'].map((t) => enqueue(b, newGoal(b), { title: t }).runId); for (const runId of aRuns) expect(wdb.updateRunStatus(runId, 'queued', 'ready')).toBe(1); expect(wdb.updateRunStatus(bRuns[0], 'queued', 'ready')).toBe(1); expect(() => wdb.updateRunStatus(bRuns[1], 'queued', 'ready')).toThrow(wdb.RunSlotBusyError); expect(wdb.countActiveRootRuns(a)).toBe(2); expect(wdb.countActiveRootRuns(b)).toBe(1); }); it('一次 tick 把空坑填满,不是一条一条来(max=3 → 三条同时起跑)', async () => { const companyId = setMaxParallel(nextCompany(), 3); const goalId = newGoal(companyId); hangingExecutor(); const tasks = ['a', 'b', 'c', 'd'].map((title) => enqueue(companyId, goalId, { title })); const res = scheduler.tick(companyId); expect(res.max).toBe(3); expect(res.started.map((s) => s.taskId)).toEqual(tasks.slice(0, 3).map((t) => t.taskId)); await flush(); expect(wdb.countActiveRootRuns(companyId)).toBe(3); expect(wdb.getTask(tasks[3].taskId).status).toBe('queued'); // 第 4 条还在队里 // 再 tick 一次不该超发:坑满就原地返回 expect(scheduler.tick(companyId).skipped).toBe('no-slot'); }); it('候选里有一条起不来时,不该白占一个名额 —— 剩下的坑照样被填满', () => { const companyId = setMaxParallel(nextCompany(), 2); const goalId = newGoal(companyId); hangingExecutor(); // 队首这条的 run 行被抹掉 = 入队做了一半(5a 那条分支),它必须被跳过而不是吃掉一个坑 const broken = enqueue(companyId, goalId, { title: 'broken' }); wdb.getDb().prepare(`UPDATE tasks SET current_run_id = NULL WHERE id = ?`).run(broken.taskId); const ok = ['a', 'b'].map((title) => enqueue(companyId, goalId, { title })); const res = scheduler.tick(companyId); // 批 3 的 `.slice(0, max - active)` 会把 broken 算进名额,于是只起来一条 expect(res.started.map((s) => s.taskId)).toEqual(ok.map((t) => t.taskId)); expect(wdb.getTask(broken.taskId).status).toBe('blocked'); }); }); it('advanceOne 挡住 queued run:不推进、不报错、节点一个都不点亮', async () => { const companyId = nextCompany(); const { runId } = enqueue(companyId, newGoal(companyId), { title: 'a' }); const calls = hangingExecutor(); const res = await engine.advanceOne(runId); expect(res.reason).toBe('run-not-running'); expect(calls).toEqual([]); // queued run 有一整套 pending 节点行(createRun 建行时就插全了)—— // 「靠没有节点行自然报错」那道防线不存在,闸门只有 run_status 这一句 const nodes = wdb.getRunState(runId).nodes; expect(nodes.length).toBeGreaterThan(0); expect(nodes.every((n) => n.status === 'pending')).toBe(true); expect(wdb.getRun(runId).run_status).toBe('queued'); }); it('startRun 拒绝 queued run(出队顺序不能反过来),markRunReady 也不再假装成功', () => { const companyId = nextCompany(); const { runId } = enqueue(companyId, newGoal(companyId), { title: 'a' }); expect(() => engine.startRun(runId)).toThrow(/还在排队中/); // markRunReady 的 from 写死成 planning,对 queued run 是空转 —— 以前它照样回 ok:'ready' expect(() => engine.markRunReady(runId)).toThrow(/未生效/); expect(wdb.getRun(runId).run_status).toBe('queued'); }); }); // ────────────────────────────────────────────────────────────── // §四.3 末:排队期间 replan // ────────────────────────────────────────────────────────────── it('验收 15:排队期间图被改,出队前 syncRunNodesToDef 补齐新节点、删掉孤儿', () => { const companyId = nextCompany(); const graphId = buildGraph(companyId, ['n1', 'n2']); const { runId } = enqueue(companyId, newGoal(companyId), { title: 'a', graphId }); expect(wdb.getRunState(runId).nodes.map((n) => n.name).sort()).toEqual(['n1', 'n2']); // 「重新规划组织」:删掉 n2、加上 n3(定义层直接改,模拟 replan 落库后的结果)。 // 关键在于 defNodeId 是节点**行 id**,所以这必须是真的删行 + 插行,改个名字不算。 ensureAgents(['n3']); const db = wdb.getDb(); const now = new Date().toISOString(); 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); const n1 = db.prepare(`SELECT id FROM work_graph_nodes WHERE work_graph_id = ? AND node_name = 'n1'`).get(graphId); const n3 = wdb.newId('wgn'); db.prepare(`INSERT INTO work_graph_nodes (id, work_graph_id, node_name, system_agent_id, node_kind, created_at) VALUES (?, ?, 'n3', 'n3', 'agent_call', ?)`).run(n3, graphId, now); db.prepare(`INSERT INTO work_graph_edges (id, work_graph_id, from_node_id, to_node_id, condition, edge_kind, created_at) VALUES (?, ?, ?, ?, 'success', 'forward', ?)`).run(wdb.newId('wge'), graphId, n1.id, n3, now); const diff = wdb.syncRunNodesToDef(runId); expect(diff.added.length).toBe(1); expect(diff.removed.length).toBe(1); expect(diff.kept).toEqual([]); const names = wdb.getRunState(runId).nodes.map((n) => n.name).sort(); expect(names).toEqual(['n1', 'n3']); expect(wdb.getRunState(runId).nodes.every((n) => n.status === 'pending')).toBe(true); }); // ────────────────────────────────────────────────────────────── // §五.1 出队 // ────────────────────────────────────────────────────────────── it('tick 一次只放一条出队(max=1),其余照排,位次按 priority → seq → created_at', () => { const companyId = nextCompany(); const goalId = newGoal(companyId); hangingExecutor(); const low = enqueue(companyId, goalId, { title: 'low', priority: 'P2' }); const high = enqueue(companyId, goalId, { title: 'high', priority: 'P0' }); // 位次口径只有调度器一处 expect(scheduler.listQueue(companyId).map((t) => t.id)).toEqual([high.taskId, low.taskId]); expect(scheduler.queuePosition(companyId, low.taskId)).toBe(2); const res = scheduler.tick(companyId); expect(res.started.map((s) => s.taskId)).toEqual([high.taskId]); expect(wdb.getTask(high.taskId).status).toBe('running'); expect(wdb.getTask(high.taskId).attempt).toBe(1); expect(wdb.getRun(high.runId).run_status).toBe('running'); // 后面那条一动不动 expect(wdb.getTask(low.taskId).status).toBe('queued'); expect(wdb.getRun(low.runId).run_status).toBe('queued'); // 坑满了,再 tick 也不会超发 expect(scheduler.tick(companyId).skipped).toBe('no-slot'); }); it('manual 任务不参与自动出队(§五.2bis:它会占着坑停在第一个节点上堵死队头)', () => { const companyId = nextCompany(); const goalId = newGoal(companyId); hangingExecutor(); const m = enqueue(companyId, goalId, { title: 'm', mode: 'manual' }); const res = scheduler.tick(companyId); expect(res.started).toEqual([]); expect(wdb.getTask(m.taskId).status).toBe('queued'); // 但它照样占位次(界面上要显示「排队 #1」) expect(scheduler.queuePosition(companyId, m.taskId)).toBe(1); }); it('验收 13:停止即出队 —— stopRun 不走 settleRun,靠 transitionRun 漏斗钩子把下一条拉起来', async () => { const companyId = nextCompany(); const goalId = newGoal(companyId); hangingExecutor(); const a = enqueue(companyId, goalId, { title: 'a', priority: 'P0' }); const b = enqueue(companyId, goalId, { title: 'b', priority: 'P1' }); scheduler.tick(companyId); expect(wdb.getTask(a.taskId).status).toBe('running'); expect(wdb.getTask(b.taskId).status).toBe('queued'); engine.stopRun(a.runId); // 这条路径全程没有 settleRun await flush(); expect(wdb.getTask(a.taskId).status).toBe('cancelled'); expect(wdb.getTask(b.taskId).status).toBe('running'); expect(wdb.getRun(b.runId).run_status).toBe('running'); }); it('验收 2:上一条正常收敛后,下一条自动起跑(无人工干预)', async () => { const companyId = nextCompany(); const goalId = newGoal(companyId); hangingExecutor(); const a = enqueue(companyId, goalId, { title: 'a', priority: 'P0' }); const b = enqueue(companyId, goalId, { title: 'b' }); scheduler.tick(companyId); // 直接推唯一漏斗,等价于 auto driver 跑完那一跳 expect(sse.transitionRun(a.runId, 'running', 'completed', {}, { summary: '用例直推终态' })).toBeTruthy(); await flush(); expect(wdb.getTask(a.taskId).status).toBe('completed'); expect(wdb.getTask(b.taskId).status).toBe('running'); }); // ────────────────────────────────────────────────────────────── // §五.1.2 依赖推导 // ────────────────────────────────────────────────────────────── it('验收 23:上游失败 → 下游 blocked 且带 upstream_failed:;上游恢复后自己回 queued', () => { const companyId = nextCompany(); const goalId = newGoal(companyId); hangingExecutor(); const a = enqueue(companyId, goalId, { title: 'a', priority: 'P0' }); const b = enqueue(companyId, goalId, { title: 'b', dependsOn: [] }); const c = enqueue(companyId, goalId, { title: 'c', dependsOn: [] }); // 依赖在 queued 状态也允许改(§六.3):B 等 A,C 等 B —— 级联要一次算到底 wdb.updateTask(b.taskId, { dependsOn: [a.taskId] }); wdb.updateTask(c.taskId, { dependsOn: [b.taskId] }); wdb.updateTask(a.taskId, { status: 'failed' }); scheduler.tick(companyId); expect(wdb.getTask(b.taskId).status).toBe('blocked'); expect(wdb.getTask(b.taskId).blocked_reason).toBe(`upstream_failed:${a.taskId}`); expect(wdb.getTask(c.taskId).status).toBe('blocked'); expect(wdb.getTask(c.taskId).blocked_reason).toBe(`upstream_failed:${b.taskId}`); // A 重跑成功 wdb.updateTask(a.taskId, { status: 'completed' }); const res = scheduler.tick(companyId); expect(res.requeued).toContain(b.taskId); expect(res.requeued).toContain(c.taskId); expect(wdb.getTask(c.taskId).status).toBe('queued'); expect(wdb.getTask(c.taskId).blocked_reason).toBeNull(); // B 依赖满足了就该被拉起来,C 还得等 B expect(res.started.map((s) => s.taskId)).toEqual([b.taskId]); }); it('别的来源的 blocked(预算耗尽那种)绝不被自动重排', () => { const companyId = nextCompany(); const goalId = newGoal(companyId); const t = enqueue(companyId, goalId, { title: 'x' }); wdb.updateTask(t.taskId, { status: 'blocked', blockedReason: 'budget-exhausted' }); const res = scheduler.tick(companyId); expect(res.requeued).toEqual([]); expect(wdb.getTask(t.taskId).status).toBe('blocked'); }); it('queued 但没有 run 行的任务不许假装在排队:标 blocked 并写明原因(且不会被自动重排)', () => { const companyId = nextCompany(); const goalId = newGoal(companyId); const t = enqueue(companyId, goalId, { title: 'x' }); wdb.getDb().prepare(`DELETE FROM graph_run_nodes WHERE run_id = ?`).run(t.runId); wdb.getDb().prepare(`DELETE FROM graph_runs WHERE id = ?`).run(t.runId); scheduler.tick(companyId); const row = wdb.getTask(t.taskId); expect(row.status).toBe('blocked'); expect(row.blocked_reason).toMatch(/^enqueue_incomplete:missing_run/); expect(scheduler.tick(companyId).requeued).toEqual([]); }); it('tick 可重入且不 throw:它挂在状态转换的尾巴上,抛出去会把一次正常收敛也带崩', () => { expect(() => scheduler.tick(undefined)).not.toThrow(); expect(scheduler.tick('no-such-company').started).toEqual([]); }); });