task-dispatch-store.test.js
7.33 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
/**
* 批 5 第 3 步：`/task/dispatch` 的存储从 system/tasklog/*.json 换成 tasks 表。
*
* **这一步唯一不能坏的是契约**：`/task/dispatch` 与 `/task/report` 的 URL、请求体、响应体
* 印在 `packages/agent-core/base/common/tools-base.md` 里，是所有 agent 的工具说明 ——
* 改一个字等于同时改所有 agent 的提示词。所以这份用例盯的是「形状没变」，不是「实现对不对」。
*/
import { describe, it, expect, beforeAll, afterAll } from 'vitest';
import { createRequire } from 'module';
import fs from 'fs';
import os from 'os';
import path from 'path';
const require_ = createRequire(import.meta.url);
describe('派单存储换成 tasks 表', () => {
let wdb, app, srv, base;
beforeAll(async () => {
process.env.GRAPH_DB_PATH = path.join(fs.mkdtempSync(path.join(os.tmpdir(), 'dispatch-store-')), 'graph.sqlite');
for (const m of ['../../services/work-graph-db', '../../services/graph-db', '../../routes/agent-chat']) {
delete require_.cache[require_.resolve(m)];
}
wdb = require_('../../services/work-graph-db');
const express = require_('express');
app = express();
app.use(express.json());
app.use('/api/agent', require_('../../routes/agent-chat'));
await new Promise((r) => { srv = app.listen(0, '127.0.0.1', r); });
base = `http://127.0.0.1:${srv.address().port}`;
});
afterAll(() => { try { srv?.close(); } catch {} });
const post = async (p, body) => {
const r = await fetch(base + p, {
method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(body),
});
return { status: r.status, body: await r.json().catch(() => ({})) };
};
/** 用真实存在的两个本体（端点会校验目录存在） */
const FROM = 'ceo';
const TO = 'qa-judge';
it('派单落进 tasks 表，且响应体是老形态（agent 提示词里印着它）', async () => {
const r = await post('/api/agent/task/dispatch', {
from: FROM, to: TO, task: '守卫用例：形态检查', priority: 'P1',
goalId: 'O-900', nodeId: 'O-900.1.1', autoExecute: false,
});
expect(r.status).toBe(200);
// 响应体的字段一个都不能少 —— 少了就是改了所有 agent 的工具说明
for (const k of ['success', 'taskId', 'from', 'to', 'task', 'priority', 'autoExecute', 'message']) {
expect(r.body).toHaveProperty(k);
}
expect(r.body.success).toBe(true);
const row = wdb.getDb().prepare('SELECT * FROM tasks WHERE id = ?').get(r.body.taskId);
expect(row).toBeTruthy();
expect(row.executor).toBe('agent'); // 调度器据此跳过它（图任务才排队）
expect(row.assignee_agent_id).toBe(TO);
expect(row.created_by).toBe(FROM);
expect(row.status).toBe('running'); // meta 的 dispatched ↔ 表的 running
// 每条派单自带一个 goal：goal_id 是 NOT NULL 真外键，派单没有「人提的需求」那一层
const goal = wdb.getDb().prepare('SELECT * FROM goals WHERE id = ?').get(row.goal_id);
expect(goal.title).toContain(FROM);
expect(goal.okr_node_id).toBe('O-900.1.1'); // objectiveId 由它的首段推导，不单独存
});
it('去重仍然拦得住，且 409 的响应体形态没变', async () => {
const body = { from: FROM, to: TO, task: '守卫用例：去重', autoExecute: false };
const first = await post('/api/agent/task/dispatch', body);
expect(first.status).toBe(200);
const dup = await post('/api/agent/task/dispatch', body);
expect(dup.status).toBe(409);
for (const k of ['success', 'error', 'existingTaskId', 'existingTask', 'message']) {
expect(dup.body).toHaveProperty(k);
}
expect(dup.body.existingTaskId).toBe(first.body.taskId);
});
it('汇报回路：状态 / summary / 完成时间都落到表上', async () => {
const d = await post('/api/agent/task/dispatch', {
from: FROM, to: TO, task: '守卫用例：汇报回路', autoExecute: false,
});
const rep = await post('/api/agent/task/report', {
taskId: d.body.taskId, from: TO, to: FROM, status: 'done', summary: '干完了',
});
expect(rep.status).toBe(200);
const row = wdb.getDb().prepare('SELECT * FROM tasks WHERE id = ?').get(d.body.taskId);
expect(row.status).toBe('completed'); // meta 的 done ↔ 表的 completed
expect(row.summary).toBe('干完了');
expect(row.finished_at).toBeTruthy();
});
it('派单不被调度器碰 —— 它没有 run 行，出队会把它误收成 cancelled', async () => {
for (const m of ['../../services/task-scheduler']) delete require_.cache[require_.resolve(m)];
const scheduler = require_('../../services/task-scheduler');
const d = await post('/api/agent/task/dispatch', {
from: FROM, to: TO, task: '守卫用例：调度器别碰我', autoExecute: false,
});
const taskId = d.body.taskId;
const row0 = wdb.getDb().prepare('SELECT * FROM tasks WHERE id = ?').get(taskId);
expect(row0.executor).toBe('agent');
// 把它摆成「最像可出队」的样子：queued + auto + 无依赖。只有 executor 这一维拦得住它。
wdb.getDb().prepare("UPDATE tasks SET status='queued', execution_mode='auto' WHERE id=?").run(taskId);
expect(scheduler.listQueue(row0.company_id).map((t) => t.id)).not.toContain(taskId);
scheduler.tick(row0.company_id);
scheduler.sweep();
// 关键断言：没有被出队、更没有被 abandonRun 收成 cancelled
const row1 = wdb.getDb().prepare('SELECT * FROM tasks WHERE id = ?').get(taskId);
expect(row1.status).toBe('queued');
expect(row1.current_run_id).toBeFalsy();
});
// 曾经的缺陷：report 的自唤醒回调写死 port 3010，跑一次测试就往线上发一条 CEO chat、
// 起一个常驻 claude CLI 会话、并留下 system/agents/ceo/_global/callback-T-*。
it('测试环境下汇报不发对外 HTTP，也不新建 callback 目录', async () => {
const http = require_('http');
const orig = http.request;
const calls = [];
http.request = function (...args) { calls.push(args[0]); return orig.apply(this, args); };
const ceoState = path.resolve(process.cwd(), 'system', 'agents', 'ceo', '_global');
const cbBefore = fs.existsSync(ceoState)
? fs.readdirSync(ceoState).filter((n) => n.startsWith('callback-')) : [];
try {
const d = await post('/api/agent/task/dispatch', {
from: FROM, to: TO, task: '守卫用例：回调不外发', autoExecute: false,
});
await post('/api/agent/task/report', {
taskId: d.body.taskId, from: TO, to: FROM, status: 'done', summary: '干完了',
});
await new Promise((r) => setTimeout(r, 200)); // 回调走 setImmediate，给它跑的机会
expect(calls).toEqual([]);
} finally {
http.request = orig;
}
const cbAfter = fs.existsSync(ceoState)
? fs.readdirSync(ceoState).filter((n) => n.startsWith('callback-')) : [];
expect(cbAfter).toEqual(cbBefore);
});
it('不再往 system/tasklog/ 写新文件', async () => {
const dir = path.resolve(process.cwd(), 'system', 'tasklog');
const before = fs.existsSync(dir) ? fs.readdirSync(dir).length : 0;
await post('/api/agent/task/dispatch', { from: FROM, to: TO, task: '守卫用例：不写文件', autoExecute: false });
const after = fs.existsSync(dir) ? fs.readdirSync(dir).length : 0;
expect(after).toBe(before);
});
});