/** * 批 3 · 排队 —— §八 验收清单的单测投影（27-P3C） * * 与 `task-scheduler.test.js` 的分工：那份钉的是**单个原语**的形状（准入守卫的判据、 * CAS 的 WHERE、唯一索引的取值域）；这一份钉的是**端到端的那句话**—— * 「人不再干预，队伍自己往前走」。所以这里几乎不直接调 `updateRunStatus` / `blockQueuedTask`， * 一律推真实入口（`engine.stopRun` / 真的烧完预算 / `sse.transitionRun` 这个唯一漏斗）， * 然后只看结果：下一条起没起来。 * * 覆盖的验收条目（编号同 §八）： * 2 T2/T3 依赖 T1 → T1 收敛后 T2 自动起跑，全程无人工干预 * 3 max=1 下第二条排队而不是 409 * 7 取消 queued → run queued→stopped，后面的顶上来 * 8 failed 的 task rerun → attempt+1、新 run、旧 run 仍在历史里 * 9 进程重启后 sweep 把队列捡起来 * 13 stopRun（不走 settleRun）后队列自动前进，不靠 60s 兜底 * 14 预算耗尽 → blocked → 同样自动前进 * 15 排队期间 replan → 出队后按新图跑（tick 里真的调了 syncRunNodesToDef） * 16 manual 不被自动拉起、不堵队；人点「开始」才占坑 * 23 上游 failed → 下游 blocked:upstream_failed；上游 rerun 成功后自动回 queued 并起跑 * ── 唯一索引的 WHERE 里没有 queued（出队原子性的全部依赖） * * 隔离手法与 graph-engine.test.js / task-scheduler.test.js 一致：**先设 env 再 require** * （graph-run-sse 在模块加载时快照 GRAPH_RUNS_DIR，先 require 再设会往仓库 data/companies/ 写 * 而且不报错）。DB 路径与那两份**故意不同**：vitest 默认按文件并行，共用一个 sqlite 文件 * 会让两份用例互相 DELETE 对方的行。 */ import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; import fs from 'fs'; import os from 'os'; import path from 'path'; describe('批 3 排队 · §八 验收', () => { let wdb; let engine; let sse; let scheduler; let validation; let runsDir; const TEST_DB_PATH = '/tmp/test-task-queue-acceptance.sqlite'; 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-queue-acc-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', '../../services/goal-validation', ]) { 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'); validation = require('../../services/goal-validation'); // goal 级验收默认会**真起 Claude CLI**（§四.10）。本文件有几条用例会让一个 goal 下的 // 多条 task 全部 completed，那正是它的触发条件 —— 不注入就会在单测里起子进程。 validation.setValidateGoal(async () => ({ pass: true, verdict: '用例注入：pass', tokens: 0 })); }); /** * `advanceOne` 的 ensureProjectRoot 会在**仓库的** data/companies/ 下给每家公司建项目根 * （它不认 GRAPH_RUNS_DIR）。只删本文件造的那些前缀目录，且必须确认里面没有 company.json * —— 那是真公司，删了就是删用户数据。 */ function cleanupLeakedCompanyDirs() { const root = path.resolve(__dirname, '../../../data/companies'); if (!fs.existsSync(root)) return; for (const cid of fs.readdirSync(root)) { if (!cid.startsWith('qacc-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 () => { // 等一拍再清：末尾那条用例的 setImmediate 钩子可能还没出栈， // 它一跑 advanceOne 就会经 ensureProjectRoot 把刚删掉的公司目录原样建回来 await new Promise((r) => setTimeout(r, 120)); scheduler.stopSweeper(); validation.setValidateGoal(null); 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); // 每条用例都从「没有兜底定时器」开始：本文件要证明的正是「队列靠漏斗钩子前进， // 不是靠 60s 扫描」。定时器开着的话，验收 13 / 14 就算钩子漏了也可能蒙混过关。 scheduler.stopSweeper(); }); // ── 夹具 ────────────────────────────────────────────────────── let seq = 0; /** 每个用例一家独立公司：唯一索引是按 company 约束的 */ const nextCompany = () => `qacc-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); } /** 建一张链式 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; } /** * 「重新规划组织」落库后的样子：把这张图的节点整批换掉。 * * 必须是真的删行 + 插行，改个 node_name 不算 —— `graph_run_nodes.def_node_id` 认的是 * 节点**行 id**，只有换行才会造出「run 上的节点行指向已不存在的定义」这种孤儿， * 而那正是 `syncRunNodesToDef` 要修的东西。 */ function rewriteGraph(graphId, names) { ensureAgents(names); 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 = ?`).run(graphId); const ids = names.map((name) => { const id = wdb.newId('wgn'); db.prepare(`INSERT INTO work_graph_nodes (id, work_graph_id, node_name, system_agent_id, node_kind, created_at) VALUES (?, ?, ?, ?, 'agent_call', ?)`).run(id, graphId, name, name, now); return id; }); for (let i = 1; i < ids.length; i++) { 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, ids[i - 1], ids[i], now); } } const newGoal = (companyId) => wdb.createGoal({ companyId, title: 'g', description: 'g' }).id; /** * 入队 —— 端点还没接上（`company.js` 里 `enqueueTaskRun` 目前零调用方），这里手工拼出 * 它承诺的那三步：建 task 行 → 建 **`runStatus:'queued'`** 的 root run（不占坑） * → task 翻 queued + 挂 current_run_id。顺序与 `enqueueTaskRun` 逐字一致 * （先 run 后 task：反过来的话中间窗口里的 tick 会把一次正常入队判成 enqueue_incomplete）。 */ function enqueue(companyId, goalId, { title, mode = 'auto', dependsOn = [], priority = 'P1', graphId, budget } = {}) { 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', budget, }); wdb.updateTask(task.id, { status: 'queued', currentRunId: run.id, queuedAt: new Date().toISOString(), }); return { taskId: task.id, runId: run.id, graphId: run.graphId }; } /** * 任务级重跑（§六.3 的 `POST /:cid/tasks/:taskId/rerun`，端点未落地时的等价三步）。 * **新起一条 run**，旧 run 一个字段都不动 —— 旧 run 留在历史里可点开正是验收 8 的要求。 */ function rerun(companyId, taskRow, { graphId } = {}) { const run = wdb.createRun({ graphKind: 'work', graphId: graphId || buildGraph(companyId), companyId, goalId: taskRow.goal_id, taskId: taskRow.id, executionMode: taskRow.execution_mode, runStatus: 'queued', }); wdb.updateTask(taskRow.id, { status: 'queued', currentRunId: run.id, startedAt: null, finishedAt: null, blockedReason: null, }); return run.id; } /** * 人在控制台上点「开始」—— `company.js:promoteQueuedTask` 的四步，逐字同序。 * manual 任务只有这一条起跑路径（调度器**故意**不碰它们，§五.2bis）。 */ function humanStart(taskRow) { const runId = taskRow.current_run_id; wdb.syncRunNodesToDef(runId); if (!sse.transitionRun(runId, 'queued', 'ready', {}, { summary: '人工出队', task_id: taskRow.id })) return null; if (!wdb.claimQueuedTask(taskRow.id, runId)) return null; return engine.startRun(runId, { mode: taskRow.execution_mode, task: taskRow.description }); } /** * 永不 settle 的 executor：让 run 稳稳停在 running，好观察队列。 * * **故意不留放行钩子**。放行会让 driver 在用例结束后继续跑一轮 advanceOne，而那时 * beforeEach 已经把 graph_run_nodes 清空了 —— 于是每条用例收尾都刷一行 * 「auto 驱动异常: graph_run_nodes 不存在」。悬着的 promise 不会拖住进程退出， * 也不会再碰库，比放行干净。 */ function hangingExecutor() { const calls = []; engine.setAgentExecutor((ctx) => { calls.push(ctx.nodeName); return new Promise(() => {}); }); return calls; } /** * 等一个条件成立。**不用固定次数的 flush**：从「漏斗钩子」到「下一条真的 running」中间 * 隔着 setImmediate + 一整条 async driver，拍脑袋数几拍出来的用例是不稳的。 * 超时就把最后一次的实际值一起报出来，免得只看到一句 timeout。 */ async function waitFor(fn, label, timeoutMs = 2000) { const deadline = Date.now() + timeoutMs; let last; for (;;) { last = fn(); if (last) return last; if (Date.now() > deadline) throw new Error(`等待超时：${label}（最后一次取到 ${JSON.stringify(last)}）`); await new Promise((r) => setImmediate(r)); } } const taskStatus = (id) => wdb.getTask(id).status; const runStatus = (id) => wdb.getRun(id).run_status; // ────────────────────────────────────────────────────────────── // 批 4：唯一索引已 DROP —— 事务准入是唯一的守卫 // ────────────────────────────────────────────────────────────── /** * 批 3 这一条钉的是「唯一索引的 WHERE 里没有 queued」（当时出队原子性靠它兜底）。 * 批 4 把索引 DROP 了 —— 但要钉的事情**没变**：超发必须拦得住。所以这一条整体反转： * · 索引必须**不存在**（`CREATE ... IF NOT EXISTS` 对存量库不生效，所以必须显式 DROP。 * 它要是还在，max>1 的第二条准入会在事务放行之后吃一句 SQLite 原文的 500）； * · 主守卫必须**真的**是主守卫，而且此刻就在拦超发（下面直接演示）。 * 断言索引不在，而不只是断言 JS 常量没了：两者一旦分叉，只看常量的用例照样绿。 */ it('唯一索引已 DROP，且主守卫独自拦得住超发 —— 删兜底之后 max=1 的行为一个字不变', () => { const row = wdb.getDb() .prepare(`SELECT sql FROM sqlite_master WHERE type = 'index' AND name = 'idx_graph_runs_one_active_root'`) .get(); expect(row).toBeUndefined(); // 取值域仍是准入判据的唯一持有者（queued 不在里面 = 排队不占坑） expect(wdb.ACTIVE_ROOT_RUN_STATUS).toEqual(['ready', 'running', 'paused']); expect(wdb.ACTIVE_ROOT_RUN_STATUS).not.toContain('queued'); // 没配 max_parallel_tasks 的公司回落 1 —— 存量行为不变的那条保证 const companyId = nextCompany(); expect(wdb.maxActiveRootRuns(companyId)).toBe(1); // 主守卫单独上阵：第一条进坑，第二条被事务拒（RunSlotBusyError，不是 SqliteError） const goalId = newGoal(companyId); const a = enqueue(companyId, goalId, { title: 'a' }); const b = enqueue(companyId, goalId, { title: 'b' }); expect(wdb.updateRunStatus(a.runId, 'queued', 'ready')).toBe(1); expect(() => wdb.updateRunStatus(b.runId, 'queued', 'ready')).toThrow(wdb.RunSlotBusyError); expect(wdb.getRun(b.runId).run_status).toBe('queued'); // 拒绝是干净的：一个字都没写 expect(wdb.countActiveRootRuns(companyId)).toBe(1); }); // ────────────────────────────────────────────────────────────── // 验收 3：第二条排队而不是 409 // ────────────────────────────────────────────────────────────── it('验收 3：max=1 时第二条任务入队不报错（这就是 409 消失的地方），两条都躺在队里各占位次', () => { const companyId = nextCompany(); const goalId = newGoal(companyId); hangingExecutor(); const a = enqueue(companyId, goalId, { title: 'a' }); const b = enqueue(companyId, goalId, { title: 'b' }); // 建两条 queued root run 本身就不该撞唯一索引 —— 旧代码正是在这一步 409 的 expect(runStatus(a.runId)).toBe('queued'); expect(runStatus(b.runId)).toBe('queued'); expect(scheduler.queuePosition(companyId, a.taskId)).toBe(1); expect(scheduler.queuePosition(companyId, b.taskId)).toBe(2); scheduler.tick(companyId); expect(taskStatus(a.taskId)).toBe('running'); expect(taskStatus(b.taskId)).toBe('queued'); // 已经有一条在跑时**再入队第三条**：这才是「下达第二个任务」的真实时序。 // 旧代码在这里抛 UNIQUE constraint → 路由翻 409；现在必须一声不响地排上。 let c; expect(() => { c = enqueue(companyId, goalId, { title: 'c' }); }).not.toThrow(); expect(runStatus(c.runId)).toBe('queued'); expect(wdb.countActiveRootRuns(companyId)).toBe(1); // 排队的一条都不占坑 // ⚠️ 位次口径：listQueue 只数 `status='queued'` 的行，跑着的那条不在队里。 // 所以此刻 b 是第 1、c 是第 2 —— §八.3 写的「queuePosition=2」是把正在跑的那条也数了进去。 // 这里钉的是**实现现在的口径**（端点回的就是这个数），口径分歧登记在返回结论里。 expect(scheduler.queuePosition(companyId, b.taskId)).toBe(1); expect(scheduler.queuePosition(companyId, c.taskId)).toBe(2); }); // ────────────────────────────────────────────────────────────── // 验收 2：依赖链自动前进 // ────────────────────────────────────────────────────────────── it('验收 2：T2/T3 依赖 T1 → T1 跑时两条都 queued；T1 收敛后 T2 自动起跑，全程无人工干预', async () => { const companyId = nextCompany(); const goalId = newGoal(companyId); hangingExecutor(); const t1 = enqueue(companyId, goalId, { title: 't1' }); const t2 = enqueue(companyId, goalId, { title: 't2' }); const t3 = enqueue(companyId, goalId, { title: 't3' }); wdb.updateTask(t2.taskId, { dependsOn: [t1.taskId] }); wdb.updateTask(t3.taskId, { dependsOn: [t1.taskId] }); // 唯一的一次人工干预就是这一下（等价于用户点「下达」）；之后一次 tick 都不许再调 scheduler.tick(companyId); expect(taskStatus(t1.taskId)).toBe('running'); expect(taskStatus(t2.taskId)).toBe('queued'); expect(taskStatus(t3.taskId)).toBe('queued'); // 依赖未满足的照样在队里占位次（界面上要显示「排队 #1 / #2」），只是不够格出队 expect(scheduler.queuePosition(companyId, t2.taskId)).toBe(1); expect(scheduler.queuePosition(companyId, t3.taskId)).toBe(2); // T1 正常收敛：推唯一漏斗，等价于 auto driver 跑完最后一个节点那一跳 expect(sse.transitionRun(t1.runId, 'running', 'completed', {}, { summary: '用例直推终态' })).toBeTruthy(); await waitFor(() => taskStatus(t2.taskId) === 'running', 'T1 收敛后 T2 自动起跑'); expect(taskStatus(t1.taskId)).toBe('completed'); expect(runStatus(t2.runId)).toBe('running'); // max=1：T3 依赖也满足了，但坑只有一个，它继续排 expect(taskStatus(t3.taskId)).toBe('queued'); expect(scheduler.queuePosition(companyId, t3.taskId)).toBe(1); // 再收敛 T2 → T3 接着自己起来（依赖链一路走到底，中间没有人碰过） expect(sse.transitionRun(t2.runId, 'running', 'completed', {}, { summary: '用例直推终态' })).toBeTruthy(); await waitFor(() => taskStatus(t3.taskId) === 'running', 'T2 收敛后 T3 自动起跑'); }); // ────────────────────────────────────────────────────────────── // 验收 7：取消排队中的任务 // ────────────────────────────────────────────────────────────── it('验收 7：取消一个 queued task → run queued→stopped、task cancelled，后面那条顶上来', 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' }); const c = enqueue(companyId, goalId, { title: 'c', priority: 'P2' }); scheduler.tick(companyId); expect(taskStatus(a.taskId)).toBe('running'); expect(scheduler.queuePosition(companyId, c.taskId)).toBe(2); // 取消 = `company.js:cancelQueuedRun` 那一跳（走漏斗，不裸写库）。 // task 状态**不在这里写**：由漏斗钩子上的 task-settle 按 RUN_TO_TASK_STATUS 收成 cancelled， // 「取消排队」在库里 / 时间线上 / 界面上因此是同一件事。 expect(sse.transitionRun(b.runId, 'queued', 'stopped', {}, { summary: '取消排队' })).toBeTruthy(); expect(runStatus(b.runId)).toBe('stopped'); await waitFor(() => taskStatus(b.taskId) === 'cancelled', '取消后 task 收成 cancelled'); // 后面那条顶上来：位次 2 → 1 expect(scheduler.queuePosition(companyId, c.taskId)).toBe(1); // A 让出坑之后，接班的是 C 而不是被取消的 B expect(sse.transitionRun(a.runId, 'running', 'completed', {}, { summary: '用例直推终态' })).toBeTruthy(); await waitFor(() => taskStatus(c.taskId) === 'running', 'C 顶上来起跑'); expect(runStatus(b.runId)).toBe('stopped'); // 被取消的那条一动不动 }); // ────────────────────────────────────────────────────────────── // 验收 8：任务级 rerun // ────────────────────────────────────────────────────────────── it('验收 8：failed 的 task rerun → attempt 累加、新起一条 run，旧 run 仍在历史里可点开', async () => { const companyId = nextCompany(); const goalId = newGoal(companyId); hangingExecutor(); const a = enqueue(companyId, goalId, { title: 'a' }); scheduler.tick(companyId); expect(wdb.getTask(a.taskId).attempt).toBe(1); // attempt 由 claimQueuedTask 在出队时 +1 // 第一次跑失败（走漏斗 → task-settle 回写） expect(sse.transitionRun(a.runId, 'running', 'failed', {}, { summary: '用例直推终态' })).toBeTruthy(); await waitFor(() => taskStatus(a.taskId) === 'failed', 'run 失败回写 task'); const oldRun = wdb.getRun(a.runId); const oldNodeCount = wdb.getRunState(a.runId).nodes.length; expect(oldNodeCount).toBeGreaterThan(0); // rerun：新 run + task 回 queued const newRunId = rerun(companyId, wdb.getTask(a.taskId)); expect(newRunId).not.toBe(a.runId); scheduler.tick(companyId); const fresh = wdb.getTask(a.taskId); expect(fresh.status).toBe('running'); expect(fresh.current_run_id).toBe(newRunId); expect(fresh.attempt).toBe(2); // 第二代次 expect(runStatus(newRunId)).toBe('running'); // 旧 run 一个字段都没被动过 —— 「点开历史」靠的就是它还在、节点行还在 const stillThere = wdb.getRun(a.runId); expect(stillThere.run_status).toBe('failed'); expect(stillThere.created_at).toBe(oldRun.created_at); expect(wdb.getRunState(a.runId).nodes.length).toBe(oldNodeCount); // 两条 run 都挂在同一条 task 下，历史列表里能同时看到 const roots = wdb.listCompanyRuns(companyId, { rootOnly: true }).filter((r) => r.task_id === a.taskId); expect(roots.map((r) => r.id).sort()).toEqual([a.runId, newRunId].sort()); // 进度也是按 run 分别算的，rerun 不会把旧 run 的进度算到新 run 头上 const prog = wdb.taskProgress([a.runId, newRunId]); expect(prog.has(a.runId)).toBe(true); expect(prog.has(newRunId)).toBe(true); }); // ────────────────────────────────────────────────────────────── // 验收 9：进程重启后队列被捡起来 // ────────────────────────────────────────────────────────────── it('验收 9：sweep 把「所有队里有人的公司」各推一次 —— 进程重启丢掉的唤醒靠它补回来', () => { const co1 = nextCompany(); const co2 = nextCompany(); hangingExecutor(); const a = enqueue(co1, newGoal(co1), { title: 'a' }); const b = enqueue(co2, newGoal(co2), { title: 'b' }); // 模拟「重启后没有任何钩子会被触发」：谁都没调过 tick，两条就那么躺着 expect(taskStatus(a.taskId)).toBe('queued'); expect(taskStatus(b.taskId)).toBe('queued'); const results = scheduler.sweep(); // 跨公司各起各的：坑是按 company 算的 expect(taskStatus(a.taskId)).toBe('running'); expect(taskStatus(b.taskId)).toBe('running'); expect(results.map((r) => r.companyId).sort()).toEqual([co1, co2].sort()); }); it('startSweeper 幂等且 unref —— 兜底定时器绝不该拖着进程不退出', () => { const t1 = scheduler.startSweeper(50); const t2 = scheduler.startSweeper(50); expect(t2).toBe(t1); // 重复调只留一个，否则每次热重载都多一个扫描器 expect(typeof t1.hasRef).toBe('function'); expect(t1.hasRef()).toBe(false); scheduler.stopSweeper(); // 停完再起要能起来（不能因为 sweepTimer 没清干净就永远起不了第二个） const t3 = scheduler.startSweeper(50); expect(t3).not.toBe(t1); scheduler.stopSweeper(); }); // ────────────────────────────────────────────────────────────── // 验收 13 / 14：两条不走 settleRun 的释放路径 // ────────────────────────────────────────────────────────────── it('验收 13：stopRun 不走 settleRun —— 队列靠 transitionRun 漏斗自己前进，不是等 60s 兜底', 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(taskStatus(a.taskId)).toBe('running'); // beforeEach 里已经 stopSweeper —— 这条用例全程没有任何定时器， // 队列还能前进就只可能是漏斗钩子干的（这正是「不是等 60s」的证明方式） engine.stopRun(a.runId); await waitFor(() => taskStatus(b.taskId) === 'running', 'stopRun 后 B 自动起跑'); expect(taskStatus(a.taskId)).toBe('cancelled'); expect(runStatus(a.runId)).toBe('stopped'); }); it('验收 14：预算耗尽 → root run blocked → 下一条同样自动起跑（这条路径也不走 settleRun）', async () => { const companyId = nextCompany(); const goalId = newGoal(companyId); const calls = hangingExecutor(); // 额度直接给 0：advanceOne 拾到第一个 ready 节点、真要花钱那一刻就判耗尽 const a = enqueue(companyId, goalId, { title: 'a', priority: 'P0', budget: { total_agent_calls: 0, spent: {} } }); const b = enqueue(companyId, goalId, { title: 'b', priority: 'P1' }); scheduler.tick(companyId); await waitFor(() => runStatus(a.runId) === 'blocked', '预算耗尽 → root run blocked'); // 一次 agent 都没起：预算闸门在发起调用之前 expect(calls).toEqual([]); // blocked 不是失败，是挂起等人加预算 —— task 跟着是 blocked 而不是 failed expect(taskStatus(a.taskId)).toBe('blocked'); await waitFor(() => taskStatus(b.taskId) === 'running', '预算耗尽让出坑后 B 自动起跑'); // A 这种 blocked 不带 upstream_failed: 前缀，绝不会被下一次 tick 自动重排回 queued expect(scheduler.tick(companyId).requeued).toEqual([]); expect(taskStatus(a.taskId)).toBe('blocked'); }); // ────────────────────────────────────────────────────────────── // 验收 15：排队期间 replan // ────────────────────────────────────────────────────────────── it('验收 15：排队期间图被重新规划 → 出队后按新图跑，没有孤儿节点也不卡死', async () => { const companyId = nextCompany(); const goalId = newGoal(companyId); const calls = hangingExecutor(); const graphId = buildGraph(companyId, ['old-entry']); const a = enqueue(companyId, goalId, { title: 'a', priority: 'P0' }); // 占着坑，让 b 真的排队 const b = enqueue(companyId, goalId, { title: 'b', priority: 'P1', graphId }); scheduler.tick(companyId); expect(taskStatus(b.taskId)).toBe('queued'); // 入队那一刻的节点行快照 expect(wdb.getRunState(b.runId).nodes.map((n) => n.name)).toEqual(['old-entry']); // 「重新规划组织」：整批换节点。定义是读时取最新，节点行还停在旧快照上。 rewriteGraph(graphId, ['new-entry']); // 让出坑，b 自动出队 —— tick 内部必须先 syncRunNodesToDef 再出队， // 漏了这一句：run 上只有一条指向已删定义的孤儿节点行，seedEntryNodes 点不亮任何东西， // run 会一路 running 到 settleRun 判「没有活跃节点」，任务无声无息地假收敛。 expect(sse.transitionRun(a.runId, 'running', 'completed', {}, { summary: '用例直推终态' })).toBeTruthy(); await waitFor(() => taskStatus(b.taskId) === 'running', 'b 出队起跑'); const names = wdb.getRunState(b.runId).nodes.map((n) => n.name); expect(names).toEqual(['new-entry']); // 按新图跑，旧节点行已被删干净（无孤儿） await waitFor(() => calls.includes('new-entry'), '新图的入口节点真的被点亮并调用'); expect(calls).not.toContain('old-entry'); expect(runStatus(b.runId)).toBe('running'); // 没卡死：真的停在新节点上跑着 }); // ────────────────────────────────────────────────────────────── // 验收 16：manual 不堵队 // ────────────────────────────────────────────────────────────── it('验收 16：manual 排在队头也不被自动拉起（不堵队），auto 照常起跑；人点「开始」才占坑', async () => { const companyId = nextCompany(); const goalId = newGoal(companyId); hangingExecutor(); // manual 优先级更高、排在队头 —— 它要是被自动拉起，就会占着坑停在第一个节点上等人， // 后面所有任务无限期堵死，而队列界面看不出任何异常（§五.2bis 的原话） const m = enqueue(companyId, goalId, { title: 'm', mode: 'manual', priority: 'P0' }); const a = enqueue(companyId, goalId, { title: 'a', mode: 'auto', priority: 'P1' }); scheduler.tick(companyId); expect(taskStatus(m.taskId)).toBe('queued'); // 队头的 manual 被跳过 expect(taskStatus(a.taskId)).toBe('running'); // 后面的 auto 照样起跑 = 没被堵住 expect(scheduler.queuePosition(companyId, m.taskId)).toBe(1); // 但位次照占，界面要显示 // auto 收敛让出坑，再 tick 多少次 manual 也不会自己起来 expect(sse.transitionRun(a.runId, 'running', 'completed', {}, { summary: '用例直推终态' })).toBeTruthy(); await waitFor(() => wdb.countActiveRootRuns(companyId) === 0, 'auto 收敛让出坑'); scheduler.tick(companyId); scheduler.tick(companyId); expect(taskStatus(m.taskId)).toBe('queued'); // 人点「开始」才起跑、才占坑 expect(humanStart(wdb.getTask(m.taskId))).toBeTruthy(); expect(taskStatus(m.taskId)).toBe('running'); expect(runStatus(m.runId)).toBe('running'); expect(wdb.countActiveRootRuns(companyId)).toBe(1); }); it('L1：人亲手起过的 manual 被退回队列后，坑一空自己接回来（不用再点一次）', async () => { const companyId = nextCompany(); const goalId = newGoal(companyId); hangingExecutor(); // 人亲手起跑一条 manual —— 这一步就是「授权」，started_at 在这里被写上 const m = enqueue(companyId, goalId, { title: 'm', mode: 'manual', priority: 'P0' }); expect(humanStart(wdb.getTask(m.taskId))).toBeTruthy(); expect(wdb.getTask(m.taskId).attempt).toBeGreaterThan(0); // 授权的事实记在 attempt 上 // 它撞上预算/挂起被退回队列（直接用数据层那条退路，等价于 admitOrRequeue 走到的分支） expect(sse.transitionRun(m.runId, 'running', 'blocked', {}, { summary: '用例：预算耗尽' })).toBeTruthy(); expect(wdb.requeueBlockedRootRun(m.runId)).toBeTruthy(); expect(taskStatus(m.taskId)).toBe('queued'); expect(runStatus(m.runId)).toBe('queued'); // 坑是空的 —— 批 3 那版这里会永远躺着（调度器 filter 只认 auto），人得**另外**再点一次 expect(wdb.countActiveRootRuns(companyId)).toBe(0); scheduler.tick(companyId); // 现在它自己回来了：人刚才那次点击被兑现，而不是被吞掉 expect(taskStatus(m.taskId)).toBe('running'); expect(runStatus(m.runId)).toBe('running'); }); it('L1 反面：从没被人起过的 manual 仍然不许自动出队（首次起跑的授权不能省）', () => { const companyId = nextCompany(); const goalId = newGoal(companyId); hangingExecutor(); const m = enqueue(companyId, goalId, { title: 'm', mode: 'manual', priority: 'P0' }); expect(wdb.getTask(m.taskId).attempt).toBe(0); // 从没被人起过 scheduler.tick(companyId); scheduler.tick(companyId); expect(taskStatus(m.taskId)).toBe('queued'); expect(wdb.countActiveRootRuns(companyId)).toBe(0); }); // ────────────────────────────────────────────────────────────── // 验收 23：blocked 是推导出来的，不是落死的 // ────────────────────────────────────────────────────────────── it('验收 23：上游 failed → 下游 blocked:upstream_failed；上游 rerun 成功后下游自己回 queued 并起跑', 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' }); wdb.updateTask(b.taskId, { dependsOn: [a.taskId] }); scheduler.tick(companyId); expect(taskStatus(a.taskId)).toBe('running'); // A 真的失败（走漏斗：settle 回写 + 释放坑各一次，次序由 afterRunTransition 保证） expect(sse.transitionRun(a.runId, 'running', 'failed', {}, { summary: '用例直推终态' })).toBeTruthy(); await waitFor(() => taskStatus(b.taskId) === 'blocked', '上游失败后下游被推导成 blocked'); expect(wdb.getTask(b.taskId).blocked_reason).toBe(`upstream_failed:${a.taskId}`); // 关键：B **不是**躺在队里假装排队 —— 它已经不在队列里了，界面上是标红的 blocked expect(scheduler.queuePosition(companyId, b.taskId)).toBe(0); // 重跑上游（§五.1.2 给的两条出路之一） const aRun2 = rerun(companyId, wdb.getTask(a.taskId)); scheduler.tick(companyId); expect(taskStatus(a.taskId)).toBe('running'); // ⚠️ 这一步的口径容易想反：B 在**上游刚被重跑**时就回 queued 了，不用等 A 跑成功。 // 因为「回 queued」的判据是 `deadDependency === null`（上游不再处于 failed/cancelled/blocked）， // 而不是「上游已 completed」—— 后者是**够不够格出队**的判据（dependenciesSatisfied），两者正交。 // 分开是对的：A 一重跑，B 在界面上就该从标红的 blocked 变回「排队中，等 A」， // 而不是等 A 跑完几分钟里一直标红说「上游失败了」。 expect(taskStatus(b.taskId)).toBe('queued'); expect(wdb.getTask(b.taskId).blocked_reason).toBeNull(); // 回了队列但**不许被起跑**：坑此刻被 A 占着，且 A 还没 completed expect(runStatus(b.runId)).toBe('queued'); expect(sse.transitionRun(aRun2, 'running', 'completed', {}, { summary: '用例直推终态' })).toBeTruthy(); // 上游恢复 → B 自己回 queued 并被同一轮 tick 拉起来。 // 这一整段就是「blocked 每次 tick 重算、不落成不可逆状态」的兑现证明： // 没有 requeueDerivedBlockedTask 的话，B 会永远躺在 blocked 里。 await waitFor(() => taskStatus(b.taskId) === 'running', '上游 rerun 成功后下游自动起跑'); expect(wdb.getTask(b.taskId).blocked_reason).toBeNull(); expect(runStatus(b.runId)).toBe('running'); }); });