/** * Unit Tests — 46 号文档 Phase 4:预算均分 + 父子终态汇总 + 同父子任务独立并发坑位 * * 覆盖三块新机制: * 1. `materializeDecomposedSubtasks` 把父任务预算按步数均分给每条子任务 * 2. `markTaskDecomposed` / `settleDecomposedParentStatus` 的 CAS 语义 * 3. `updateRunStatus` 里新增的「同父 decomposed_work 子任务最多 3 并行」坑位, * 与公司级 `max_parallel_tasks` 完全独立 * 4. `task-settle.settleParentTaskIfAllChildrenDone`(父子终态汇总) * * 隔离手法同 `task-decomposition-dispatch.test.js`:先设 GRAPH_DB_PATH 再 require。 */ import { describe, it, expect, beforeAll, afterAll, beforeEach, } from 'vitest'; import fs from 'fs'; const TEST_DB_PATH = '/tmp/test-task-decomp-phase4.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/graph-run-sse', '../../services/task-settle', ]; const MIN_ACCEPTANCE = { dimensions: [{ name: 'done', weight: 1, pass_threshold: 80 }] }; describe('46 号文档 Phase 4', () => { let wdb; let settle; beforeAll(() => { for (const f of DB_FILES) { try { fs.unlinkSync(f); } catch {} } process.env.GRAPH_DB_PATH = TEST_DB_PATH; for (const m of MODULES) delete require.cache[require.resolve(m)]; wdb = require('../../services/work-graph-db'); settle = require('../../services/task-settle'); }); afterAll(() => { for (const f of DB_FILES) { try { fs.unlinkSync(f); } catch {} } delete process.env.GRAPH_DB_PATH; for (const m of MODULES) delete require.cache[require.resolve(m)]; }); 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 system_agents; `); db.pragma('foreign_keys = ON'); }); let seq = 0; function nextCompany() { return `p4-co-${++seq}`; } // ── 1. 预算均分 ────────────────────────────────────────────────────── describe('materializeDecomposedSubtasks:预算均分', () => { it('父任务有 budget_json 时,按步数均分给每条子任务', () => { const companyId = nextCompany(); const goal = wdb.createGoal({ companyId, title: 'g', description: 'd' }); const parent = wdb.createTask({ goalId: goal.id, title: 'p', description: 'd', budgetJson: { total_agent_calls: 100, total_tokens: 5000000, wallclock_ms: 7200000 }, }); const steps = [ { seq: 1, title: '第一步', depends_on_hint: [], acceptance: MIN_ACCEPTANCE }, { seq: 2, title: '第二步', depends_on_hint: [], acceptance: MIN_ACCEPTANCE }, { seq: 3, title: '第三步', depends_on_hint: [], acceptance: MIN_ACCEPTANCE }, { seq: 4, title: '第四步', depends_on_hint: [], acceptance: MIN_ACCEPTANCE }, ]; const ids = wdb.materializeDecomposedSubtasks({ task: wdb.getTask(parent.id), steps }); expect(ids).toHaveLength(4); for (const id of ids) { const budget = JSON.parse(wdb.getTask(id).budget_json); expect(budget.total_agent_calls).toBe(25); // 100 / 4 expect(budget.total_tokens).toBe(1250000); // 5000000 / 4 expect(budget.wallclock_ms).toBe(1800000); // 7200000 / 4 expect(budget.spent).toEqual({ agent_calls: 0, tokens: 0 }); } }); it('父任务没有 budget_json 时,退到 DEFAULT_TASK_BUDGET 的形状再均分', () => { const companyId = nextCompany(); const goal = wdb.createGoal({ companyId, title: 'g', description: 'd' }); const parent = wdb.createTask({ goalId: goal.id, title: 'p', description: 'd' }); const steps = [ { seq: 1, title: '一步', depends_on_hint: [], acceptance: MIN_ACCEPTANCE }, { seq: 2, title: '二步', depends_on_hint: [], acceptance: MIN_ACCEPTANCE }, ]; const ids = wdb.materializeDecomposedSubtasks({ task: wdb.getTask(parent.id), steps }); expect(ids).toHaveLength(2); const budget = JSON.parse(wdb.getTask(ids[0]).budget_json); expect(budget.total_agent_calls).toBe(50); // 100(默认)/ 2 }); it('均分结果保底至少 1,不会因为步数多而算出 0', () => { const companyId = nextCompany(); const goal = wdb.createGoal({ companyId, title: 'g', description: 'd' }); const parent = wdb.createTask({ goalId: goal.id, title: 'p', description: 'd', budgetJson: { total_agent_calls: 1, total_tokens: 1, wallclock_ms: 1 }, }); const steps = [ { seq: 1, title: '一步', depends_on_hint: [], acceptance: MIN_ACCEPTANCE }, { seq: 2, title: '二步', depends_on_hint: [], acceptance: MIN_ACCEPTANCE }, { seq: 3, title: '三步', depends_on_hint: [], acceptance: MIN_ACCEPTANCE }, ]; const ids = wdb.materializeDecomposedSubtasks({ task: wdb.getTask(parent.id), steps }); for (const id of ids) { const budget = JSON.parse(wdb.getTask(id).budget_json); expect(budget.total_agent_calls).toBeGreaterThanOrEqual(1); expect(budget.total_tokens).toBeGreaterThanOrEqual(1); expect(budget.wallclock_ms).toBeGreaterThanOrEqual(1); } }); }); // ── 2. markTaskDecomposed / settleDecomposedParentStatus ──────────── describe('markTaskDecomposed / settleDecomposedParentStatus:CAS 语义', () => { it('从 queued 转:成功后 status 变 decomposed', () => { const companyId = nextCompany(); const goal = wdb.createGoal({ companyId, title: 'g', description: 'd' }); const task = wdb.createTask({ goalId: goal.id, title: 't', description: 'd' }); wdb.updateTask(task.id, { status: 'queued' }); expect(wdb.markTaskDecomposed(task.id)).toBe(true); expect(wdb.getTask(task.id).status).toBe('decomposed'); // 已经是 decomposed 了,重复调用拿不到锁 expect(wdb.markTaskDecomposed(task.id)).toBe(false); }); // 真实回归纠正过的行为:task-scheduler 抢坑几乎是瞬间的(毫秒级),而 subtask-planning // 是真实 LLM 调用(数十秒),等它拆完回来时父任务几乎必然已经是 running 而不是 queued // 了——这不是小概率分支,是最常见的那条路径,必须能从 running 接管,否则拆分结果 // 几乎每次都接管失败、白白物化出一堆没人认领的孤儿子任务(真实回归当场撞见过一次)。 it('从 running 转:真实场景里最常见的路径,也能成功接管', () => { const companyId = nextCompany(); const goal = wdb.createGoal({ companyId, title: 'g', description: 'd' }); const task = wdb.createTask({ goalId: goal.id, title: 't', description: 'd' }); wdb.updateTask(task.id, { status: 'running' }); expect(wdb.markTaskDecomposed(task.id)).toBe(true); expect(wdb.getTask(task.id).status).toBe('decomposed'); }); it('从 blocked 转:同样还没到终态,可以接管', () => { const companyId = nextCompany(); const goal = wdb.createGoal({ companyId, title: 'g', description: 'd' }); const task = wdb.createTask({ goalId: goal.id, title: 't', description: 'd' }); wdb.updateTask(task.id, { status: 'blocked' }); expect(wdb.markTaskDecomposed(task.id)).toBe(true); expect(wdb.getTask(task.id).status).toBe('decomposed'); }); it.each(['completed', 'failed', 'cancelled'])( '父任务已经是终态(%s)时天然拿不到锁(真赢了竞态的情形,不覆盖)', (terminalStatus) => { const companyId = nextCompany(); const goal = wdb.createGoal({ companyId, title: 'g', description: 'd' }); const task = wdb.createTask({ goalId: goal.id, title: 't', description: 'd' }); wdb.updateTask(task.id, { status: terminalStatus }); // 模拟父 run 已经先跑完/失败/取消了 expect(wdb.markTaskDecomposed(task.id)).toBe(false); expect(wdb.getTask(task.id).status).toBe(terminalStatus); // 没被覆盖 }, ); it('settleDecomposedParentStatus 只能从 decomposed 转终态,CAS 挡重复结算', () => { const companyId = nextCompany(); const goal = wdb.createGoal({ companyId, title: 'g', description: 'd' }); const task = wdb.createTask({ goalId: goal.id, title: 't', description: 'd' }); wdb.updateTask(task.id, { status: 'decomposed' }); expect(wdb.settleDecomposedParentStatus(task.id, 'completed')).toBe(true); expect(wdb.getTask(task.id).status).toBe('completed'); expect(wdb.getTask(task.id).finished_at).toBeTruthy(); // 已经结算过,第二次调用(哪怕传别的状态)拿不到锁,不会覆盖 expect(wdb.settleDecomposedParentStatus(task.id, 'failed')).toBe(false); expect(wdb.getTask(task.id).status).toBe('completed'); }); it('settleDecomposedParentStatus 只接受 completed/failed,别的值直接抛', () => { expect(() => wdb.settleDecomposedParentStatus('whatever-id', 'blocked')).toThrow(/只能是/); }); }); // ── 3. 同父 decomposed_work 子任务的独立并发坑位(上限 3) ────────── describe('updateRunStatus:同父 decomposed_work 子任务并发上限 3', () => { function makeChild(companyId, parentId, i) { return wdb.createTask({ goalId: wdb.getTask(parentId).goal_id, companyId, title: `子任务 ${i}`, description: 'd', parentTaskId: parentId, subtaskKind: 'decomposed_work', acceptanceContract: MIN_ACCEPTANCE, }); } it('前 3 个同父子任务的 run 都能进 ready,第 4 个撞 RunSlotBusyError(scope=decomposed-sibling)', () => { const companyId = nextCompany(); const goal = wdb.createGoal({ companyId, title: 'g', description: 'd' }); const parent = wdb.createTask({ goalId: goal.id, companyId, title: 'p', description: 'd' }); const runs = []; for (let i = 1; i <= 4; i += 1) { const child = makeChild(companyId, parent.id, i); const run = wdb.createRun({ graphKind: 'work', graphId: 'placeholder', companyId, taskId: child.id, runStatus: 'queued', }); runs.push(wdb.getRun(run.id)); } for (let i = 0; i < 3; i += 1) { expect(wdb.updateRunStatus(runs[i].id, 'queued', 'ready')).toBe(1); } let caught = null; try { wdb.updateRunStatus(runs[3].id, 'queued', 'ready'); } catch (e) { caught = e; } expect(caught).toBeInstanceOf(wdb.RunSlotBusyError); expect(caught.scope).toBe('decomposed-sibling'); expect(caught.max).toBe(wdb.DECOMPOSED_SIBLING_CONCURRENCY); expect(caught.heldBy).toHaveLength(3); // 第 4 条依旧是 queued(没被误占) expect(wdb.getRun(runs[3].id).run_status).toBe('queued'); }); it('decomposed_work 子任务的 run 不计入公司级 countActiveRootRuns', () => { const companyId = nextCompany(); const goal = wdb.createGoal({ companyId, title: 'g', description: 'd' }); const parent = wdb.createTask({ goalId: goal.id, companyId, title: 'p', description: 'd' }); const child = makeChild(companyId, parent.id, 1); const run = wdb.createRun({ graphKind: 'work', graphId: 'placeholder', companyId, taskId: child.id, runStatus: 'queued', }); wdb.updateRunStatus(run.id, 'queued', 'ready'); expect(wdb.countActiveRootRuns(companyId)).toBe(0); }); it('公司级坑位(默认 1)被普通任务占满时,decomposed_work 子任务仍能独立进坑(互不占用)', () => { const companyId = nextCompany(); const goal = wdb.createGoal({ companyId, title: 'g', description: 'd' }); // 一条普通任务占满公司级坑位(默认 max_parallel_tasks=1) const ordinary = wdb.createTask({ goalId: goal.id, companyId, title: 'o', description: 'd' }); const ordinaryRun = wdb.createRun({ graphKind: 'org', graphId: companyId, companyId, taskId: ordinary.id, runStatus: 'queued', }); expect(wdb.updateRunStatus(ordinaryRun.id, 'queued', 'ready')).toBe(1); expect(wdb.countActiveRootRuns(companyId)).toBe(1); // decomposed_work 子任务照样能进坑——不共用这个池子 const parent = wdb.createTask({ goalId: goal.id, companyId, title: 'p', description: 'd' }); const child = makeChild(companyId, parent.id, 1); const childRun = wdb.createRun({ graphKind: 'work', graphId: 'placeholder', companyId, taskId: child.id, runStatus: 'queued', }); expect(wdb.updateRunStatus(childRun.id, 'queued', 'ready')).toBe(1); }); it('不同父任务的 decomposed_work 子任务各自独立计数,互不占用彼此的 3 个坑位', () => { const companyId = nextCompany(); const goal = wdb.createGoal({ companyId, title: 'g', description: 'd' }); const parentA = wdb.createTask({ goalId: goal.id, companyId, title: 'pa', description: 'd' }); const parentB = wdb.createTask({ goalId: goal.id, companyId, title: 'pb', description: 'd' }); // parentA 占满 3 个坑位 for (let i = 1; i <= 3; i += 1) { const child = makeChild(companyId, parentA.id, i); const run = wdb.createRun({ graphKind: 'work', graphId: 'placeholder', companyId, taskId: child.id, runStatus: 'queued', }); expect(wdb.updateRunStatus(run.id, 'queued', 'ready')).toBe(1); } // parentB 的第一个子任务不受影响 const childB = makeChild(companyId, parentB.id, 1); const runB = wdb.createRun({ graphKind: 'work', graphId: 'placeholder', companyId, taskId: childB.id, runStatus: 'queued', }); expect(wdb.updateRunStatus(runB.id, 'queued', 'ready')).toBe(1); }); }); // ── 4. settleParentTaskIfAllChildrenDone(父子终态汇总) ──────────── describe('task-settle.settleParentTaskIfAllChildrenDone', () => { function setupParentWithChildren(statuses) { const companyId = nextCompany(); const goal = wdb.createGoal({ companyId, title: 'g', description: 'd' }); const parent = wdb.createTask({ goalId: goal.id, companyId, title: 'p', description: 'd' }); wdb.updateTask(parent.id, { status: 'decomposed' }); const children = statuses.map((status, i) => { const c = wdb.createTask({ goalId: goal.id, companyId, title: `c${i}`, description: 'd', parentTaskId: parent.id, subtaskKind: 'decomposed_work', acceptanceContract: MIN_ACCEPTANCE, }); wdb.updateTask(c.id, { status }); return c; }); return { companyId, goal, parent, children }; } it('还有子任务未到终态时返回 null,父任务保持 decomposed', () => { const { parent } = setupParentWithChildren(['completed', 'running']); expect(settle.settleParentTaskIfAllChildrenDone(parent.id)).toBeNull(); expect(wdb.getTask(parent.id).status).toBe('decomposed'); }); it('全部子任务 completed → 父任务结算成 completed', async () => { const { parent, goal } = setupParentWithChildren(['completed', 'completed']); expect(settle.settleParentTaskIfAllChildrenDone(parent.id)).toBe('completed'); expect(wdb.getTask(parent.id).status).toBe('completed'); // goal 收敛的 CAS 占位值也应该已经写下去(不再是 active)——见函数注释里 // "父任务转终态不会自动触发 goal 收敛,必须补一次调用" 那段 expect(wdb.getGoal(goal.id).status).not.toBe('active'); // 让可能的异步验收(若被触发)跑完,避免测试进程留下悬空的 unhandled work await new Promise((r) => setImmediate(r)); }); it('有子任务 failed → 父任务结算成 failed', async () => { const { parent } = setupParentWithChildren(['completed', 'failed']); expect(settle.settleParentTaskIfAllChildrenDone(parent.id)).toBe('failed'); expect(wdb.getTask(parent.id).status).toBe('failed'); await new Promise((r) => setImmediate(r)); }); it('没有子任务时返回 null', () => { const companyId = nextCompany(); const goal = wdb.createGoal({ companyId, title: 'g', description: 'd' }); const parent = wdb.createTask({ goalId: goal.id, companyId, title: 'p', description: 'd' }); expect(settle.settleParentTaskIfAllChildrenDone(parent.id)).toBeNull(); }); it('并发调用只有一次生效(CAS):第二次调用返回 null', async () => { const { parent } = setupParentWithChildren(['completed']); expect(settle.settleParentTaskIfAllChildrenDone(parent.id)).toBe('completed'); expect(settle.settleParentTaskIfAllChildrenDone(parent.id)).toBeNull(); await new Promise((r) => setImmediate(r)); }); }); });