Commit 71cf45e18dd30853cb081fd4c63dc4175b92ff77

Authored by yibo
1 parent 12880b86

feat(27-P3C): 批 5 第五、六步 —— 删掉会话反推,调度器只管 executor='graph'

## 第 6 步:调度器只管图任务(一行判据,但漏了是静默的)
listQueue / deriveQueueBlocks / sweep 三处都加 executor='graph'。
派单任务虽然也进 tasks 表,但它的执行载体是「直接起一次 agent 对话」,
由 /task/dispatch 自己触发。漏这一条的后果:调度器拿一条根本没有 run 行的
task 去出队 → syncRunNodesToDef 判空 → abandonRun → **把一条正在被 agent
执行的派单收成 cancelled**。守卫用例把它摆成「最像可出队」的样子
(queued + auto + 无依赖),只有 executor 这一维拦得住它。

## 第 5 步:GET /api/tasks 改读表,删掉 122 行的会话反推
原来的 getAggregatedTasks 是启发式:扫全部 agent 的 conversations,按会话 id
里的 T-xxx 归组,再用三条策略(meta 的 parentTaskId / from agent 的活跃会话
时间包含 / master conv 的标题相似)**猜**父子关系。猜出来的树没法验证对错,
而且每次请求都要遍历所有 agent 的会话目录 —— route 那 10 秒缓存就是为它加的。

现在父子关系是 tasks.parent_task_id,派单时就显式写下来了,不用猜。
task-service.js 391 行 → 236 行;_inferParentChild / _loadTaskMetas /
TASK_META_DIR 一并删除(改完就是零调用方)。

**getTaskTimeline 保留**,但来源拆清楚:任务树从表来(parent_task_id,
可验证),时间线正文仍从 agent 的 conversations/ 扫 —— 那是一手记录,
没有替代品。删掉的只是「从会话反推任务树」那一半,不是会话本身。

两处口径变化(写在代码注释里):
- agents:原来是「会话里出现过的 agent 集合」,现在是「谁派的 + 派给谁」。
  对派单任务这两者本来就是同一件事
- conversationCount:原来是扫出来的文件数,现在是「有没有起过执行会话」(0/1)。
  为一个列表字段遍历所有 agent 的会话目录,正是这次要去掉的那笔开销;
  真要看内容走 /api/tasks/:taskId/timeline,那条仍读一手记录

## 漏改一个消费方,真打接口才发现
routes/tasks.js 自己还在算 t.conversations.length,而 conversations 是会话
反推那套的产物、已随之删除 → 500「Cannot read properties of undefined」。
测试没覆盖这条路(它只在真跑接口时才走到)。已改成用 t.conversationCount。

真打验证:GET /api/tasks 返回 17 条、12 个字段齐全 —— 条数比改前多,
因为测试跑出来的派单现在也进表了,这本身就是「存储真的换了」的证据。

全量:132 文件 / 1257 测试全绿。

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
server/routes/tasks.js
... ... @@ -30,7 +30,9 @@ router.get(&#39;/&#39;, (req, res) =&gt; {
30 30 children: t.children,
31 31 isRoot: t.isRoot,
32 32 agents: t.agents,
33   - conversationCount: t.conversations.length,
  33 + // 批 5:会话数由 TaskService 直接给(原来这里自己取 t.conversations.length,
  34 + // 而 conversations 是会话反推那套的产物,已随之删除)
  35 + conversationCount: t.conversationCount,
34 36 }));
35 37  
36 38 _taskListCache = taskList;
... ...
server/services/task-scheduler.js
... ... @@ -111,12 +111,18 @@ function dependenciesSatisfied(task) {
111 111 *
112 112 * @returns {object[]} 该公司全部 queued task(含 manual 的:它们在界面上同样占位次)
113 113 */
  114 +// 批 5:只认 executor='graph'。agent 派单(executor='agent')虽然也进 tasks 表,
  115 +// 但它的执行载体是「直接起一次 agent 对话」,由 /task/dispatch 自己触发 ——
  116 +// 进表只为了可见、可查、可串成派单树,**不该被调度器出队**。
  117 +// 漏了这一条的后果是静默的:调度器会拿一条根本没有 run 行的 task 去 syncRunNodesToDef,
  118 +// 5a 判空 → abandonRun → 把一条正在被 agent 执行的派单收成 cancelled。
114 119 function listQueue(companyId) {
115 120 return wdb()
116 121 .getDb()
117 122 .prepare(`
118 123 SELECT * FROM tasks
119 124 WHERE company_id = @companyId AND status = 'queued' AND archived_at IS NULL
  125 + AND executor = 'graph'
120 126 ORDER BY priority, seq, created_at
121 127 `)
122 128 .all({ companyId });
... ... @@ -156,7 +162,10 @@ function deriveQueueBlocks(companyId) {
156 162 const requeued = [];
157 163  
158 164 for (let pass = 0; ; pass++) {
159   - const rows = listCompanyTasks(companyId).filter((t) => t.status === 'queued' || t.status === 'blocked');
  165 + // 同上:依赖推导也只管图任务。派单任务没有 depends_on 语义(它的先后靠 parent_task_id 表达),
  166 + // 放进来只会让它被判成「依赖不可能满足」而无辜转 blocked。
  167 + const rows = listCompanyTasks(companyId)
  168 + .filter((t) => t.executor === 'graph' && (t.status === 'queued' || t.status === 'blocked'));
160 169 let changed = 0;
161 170 for (const t of rows) {
162 171 const dead = deadDependency(t);
... ... @@ -373,6 +382,7 @@ function sweep() {
373 382 .prepare(`
374 383 SELECT DISTINCT company_id FROM tasks
375 384 WHERE archived_at IS NULL
  385 + AND executor = 'graph'
376 386 AND (status = 'queued'
377 387 OR (status = 'blocked' AND blocked_reason LIKE @prefix))
378 388 `)
... ...
server/services/task-service.js
... ... @@ -4,7 +4,6 @@ const locator = require(&#39;./agent-locator&#39;);
4 4 const { listCompanyAgents } = require('./company-service');
5 5  
6 6 const SYSTEM_AGENTS_DIR = require('./agent-locator').AGENTS_ROOT;
7   -const TASK_META_DIR = path.resolve(__dirname, '../../system/tasklog');
8 7  
9 8 /**
10 9 * 递归扫描并读取系统中所有 Agent 的会话,
... ... @@ -34,18 +33,6 @@ class TaskService {
34 33 return list;
35 34 }
36 35  
37   - _loadTaskMetas() {
38   - const metas = new Map();
39   - if (!fs.existsSync(TASK_META_DIR)) return metas;
40   - const files = fs.readdirSync(TASK_META_DIR).filter(f => f.endsWith('.json'));
41   - for (const f of files) {
42   - try {
43   - const tm = JSON.parse(fs.readFileSync(path.join(TASK_META_DIR, f), 'utf-8'));
44   - if (tm.taskId) metas.set(tm.taskId, tm);
45   - } catch(e) {}
46   - }
47   - return metas;
48   - }
49 36  
50 37 /**
51 38 * 从 Conversation 聚合 Task 列表,并通过多种手段推断父子关系:
... ... @@ -53,67 +40,55 @@ class TaskService {
53 40 * 2) conversationId 模式推断(task-T-xxx 从 tpl-train-* 派出,tpl-train 从 conv-* 派出)
54 41 * 3) title 中可解析的 task reference([TASK from xxx])
55 42 */
  43 + /**
  44 + * 任务列表 —— 批 5 起**读 tasks 表**,不再从会话记录反推。
  45 + *
  46 + * 原来这里是 122 行的启发式:扫全部 agent 的 conversations,按会话 id 里的 `T-xxx` 归组,
  47 + * 再用三条策略(meta 的 parentTaskId / from agent 的活跃会话时间包含 / master conv 的
  48 + * 标题相似)**猜**父子关系。猜出来的树没法验证对错,而且每次请求都要遍历所有 agent 的
  49 + * 会话目录(route 那 10 秒缓存就是为它加的)。
  50 + *
  51 + * 现在父子关系是 `tasks.parent_task_id` —— 派单时就显式写下来了,不用猜。
  52 + *
  53 + * **响应字段一个没少**(TaskTimeline.vue 按名取值)。两处口径变化写在下面。
  54 + */
56 55 getAggregatedTasks() {
57   - const convs = this.getAllConversations();
58   - const taskMetas = this._loadTaskMetas();
59   - const taskMap = new Map();
60   -
61   - // 第一遍:建立所有 task group
62   - for (const c of convs) {
63   - let taskId = this._extractTaskId(c.conversationId);
64   -
65   - if (!taskMap.has(taskId)) {
66   - const meta = taskMetas.get(taskId);
67   - taskMap.set(taskId, {
68   - taskId,
69   - title: c.meta.title || taskId,
70   - createdAt: c.meta.createdAt || new Date().toISOString(),
71   - updatedAt: c.meta.updatedAt || c.meta.createdAt,
72   - status: meta?.status || c.meta.status || 'unknown',
73   - from: meta?.from || null,
74   - to: meta?.to || null,
75   - parentTaskId: meta?.parentTaskId || null,
76   - agents: new Set(),
77   - conversations: [],
78   - children: [],
79   - });
80   - }
81   -
82   - const t = taskMap.get(taskId);
83   - t.agents.add(`${c.companyId}:${c.agentId}`);
84   - t.conversations.push(c);
85   -
86   - if (c.meta.createdAt && new Date(c.meta.createdAt) < new Date(t.createdAt)) t.createdAt = c.meta.createdAt;
87   - if (c.meta.updatedAt && new Date(c.meta.updatedAt) > new Date(t.updatedAt)) t.updatedAt = c.meta.updatedAt;
88   - }
89   -
90   - // 第二遍:推断父子关系
91   - this._inferParentChild(taskMap, taskMetas);
92   -
93   - // 第三遍:构建 children 列表
94   - for (const t of taskMap.values()) {
95   - if (t.parentTaskId && taskMap.has(t.parentTaskId)) {
96   - const parent = taskMap.get(t.parentTaskId);
97   - if (!parent.children.includes(t.taskId)) {
98   - parent.children.push(t.taskId);
99   - }
100   - }
101   - }
102   -
103   - const tasks = Array.from(taskMap.values()).map(t => ({
104   - ...t,
105   - agents: Array.from(t.agents),
106   - isRoot: !t.parentTaskId || !taskMap.has(t.parentTaskId),
107   - conversationCount: t.conversations.length,
  56 + const wdb = require('./work-graph-db');
  57 + const rows = wdb.getDb().prepare(`
  58 + SELECT * FROM tasks WHERE archived_at IS NULL ORDER BY created_at DESC
  59 + `).all();
  60 +
  61 + // 表里的取值域是 tasks 的,而这个端点历史上吐的是 meta 的(dispatched/done)。
  62 + // 前端按字符串比对状态,所以这里翻译回去 —— 换掉存储不该让界面跟着改。
  63 + const TO_META = { running: 'dispatched', completed: 'done', queued: 'dispatched', draft: 'dispatched' };
  64 + const childrenOf = new Map();
  65 + for (const r of rows) {
  66 + if (!r.parent_task_id) continue;
  67 + if (!childrenOf.has(r.parent_task_id)) childrenOf.set(r.parent_task_id, []);
  68 + childrenOf.get(r.parent_task_id).push(r.id);
  69 + }
  70 +
  71 + return rows.map((r) => ({
  72 + taskId: r.id,
  73 + title: r.title,
  74 + createdAt: r.created_at,
  75 + updatedAt: r.updated_at,
  76 + status: TO_META[r.status] || r.status,
  77 + from: r.created_by,
  78 + to: r.assignee_agent_id,
  79 + parentTaskId: r.parent_task_id,
  80 + children: childrenOf.get(r.id) || [],
  81 + isRoot: !r.parent_task_id,
  82 + // agents:原来是「会话里出现过的 agent 集合」,现在是「派给谁 + 谁派的」。
  83 + // 对派单任务这两者本来就是同一件事;图任务没有派单对象,只有 created_by。
  84 + agents: [r.created_by, r.assignee_agent_id].filter(Boolean).filter((v, i, a) => a.indexOf(v) === i),
  85 + // conversationCount:原来是扫出来的会话文件数,现在是「有没有起过执行会话」(0/1)。
  86 + // 为一个列表字段去遍历所有 agent 的会话目录,正是这次要去掉的那笔开销;
  87 + // 真要看会话内容走 /api/tasks/:taskId/timeline,那条仍读一手记录。
  88 + conversationCount: r.execution_conv_id ? 1 : 0,
108 89 }));
109   -
110   - tasks.sort((a, b) => new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime());
111   - return tasks;
112 90 }
113 91  
114   - /**
115   - * 从 conversationId 提取逻辑 taskId
116   - */
117 92 _extractTaskId(convId) {
118 93 // "task-T-xxx" / "callback-T-xxx" → "T-xxx"
119 94 if (convId.includes('-T-')) {
... ... @@ -124,142 +99,15 @@ class TaskService {
124 99 return convId;
125 100 }
126 101  
  102 +
127 103 /**
128   - * 多策略推断父子关系:
129   - *
130   - * 策略1: .tasks/ meta 中的 parentTaskId(直接有的情况)
131   - * 策略2: 通过 "from" agent 的 活跃会话 推断
132   - * - T-xxx 的 meta.from = "innovator"
133   - * - innovator 同时有一个 tpl-train-* 会话(时间上包含 T-xxx)
134   - * → tpl-train-* 就是 T-xxx 的父任务
135   - * 策略3: 通过 master conv 触发 tpl-train 推断
136   - * - conv-xxx (master) 的时间 < tpl-train-xxx 的时间 且 title 包含相同模板名
137   - * → conv-xxx 是 tpl-train-xxx 的父任务
  104 + * 某条任务(含子任务)的跨 agent 时间线。
  105 + *
  106 + * 批 5 之后这里是**两个来源合起来**,刻意分清:
  107 + * · 任务树(有哪些子任务)→ `tasks` 表的 `parent_task_id`,显式、可验证;
  108 + * · 时间线正文 → agent 的 `conversations/` 目录,那是**一手记录**,没有替代品。
  109 + * 删掉的只是「从会话反推任务树」那一半,不是会话本身。
138 110 */
139   - _inferParentChild(taskMap, taskMetas) {
140   - const allTasks = Array.from(taskMap.values());
141   -
142   - // 策略1: 已有 parentTaskId (来自 .tasks/ meta),已在构建时设置
143   -
144   - // 策略2: T-xxx 任务,推断父任务为 from agent 的活跃 tpl-train/task 会话
145   - for (const t of allTasks) {
146   - if (t.parentTaskId) continue;
147   - if (!t.taskId.startsWith('T-')) continue;
148   -
149   - const meta = taskMetas.get(t.taskId);
150   - if (!meta) continue;
151   -
152   - const fromAgent = meta.from;
153   - const taskCreatedAt = new Date(meta.createdAt);
154   -
155   - // 找 fromAgent 的其他会话,时间上在这个 task 之前开始且之后结束/仍活跃
156   - let bestParent = null;
157   - let bestTimeDiff = Infinity;
158   -
159   - for (const candidate of allTasks) {
160   - if (candidate.taskId === t.taskId) continue;
161   - if (candidate.taskId.startsWith('T-') && !taskMetas.has(candidate.taskId)) continue;
162   -
163   - const hasAgent = candidate.conversations.some(c => c.agentId === fromAgent);
164   - if (!hasAgent) continue;
165   -
166   - const candidateCreated = new Date(candidate.createdAt);
167   - const candidateUpdated = new Date(candidate.updatedAt);
168   -
169   - if (candidateCreated < taskCreatedAt && candidateUpdated >= taskCreatedAt) {
170   - const diff = taskCreatedAt - candidateCreated;
171   - if (diff < bestTimeDiff) {
172   - bestTimeDiff = diff;
173   - bestParent = candidate;
174   - }
175   - }
176   - }
177   -
178   - if (bestParent) {
179   - t.parentTaskId = bestParent.taskId;
180   - // console.log(`[TaskService] 🔗 Inferred parent: ${t.taskId} → ${bestParent.taskId}`);
181   - }
182   - }
183   -
184   - // 策略4: T-xxx from master,但 master 的 conv-* 已经结束的情况
185   - // 这种情况发生在:master 原始对话结束后,通过 callback 链路继续派发新任务
186   - // 例如 conv-xxx → tpl-train → T-a (callback → master → 新 dispatch T-b)
187   - // 此时 T-b 应该关联到 conv-xxx(同一个用户请求)
188   - for (const t of allTasks) {
189   - if (t.parentTaskId) continue;
190   - if (!t.taskId.startsWith('T-')) continue;
191   -
192   - const meta = taskMetas.get(t.taskId);
193   - if (!meta || meta.from !== 'master') continue;
194   -
195   - const taskCreatedAt = new Date(meta.createdAt);
196   -
197   - // 找最近的 master conv-* 会话,窗口放宽到 30 分钟
198   - let bestConv = null;
199   - let bestDiff = 30 * 60 * 1000;
200   -
201   - for (const candidate of allTasks) {
202   - if (!candidate.taskId.startsWith('conv-')) continue;
203   - if (!candidate.conversations.some(c => c.agentId === 'master')) continue;
204   -
205   - const cCreated = new Date(candidate.createdAt);
206   - const diff = taskCreatedAt - cCreated;
207   -
208   - if (diff > 0 && diff < bestDiff) {
209   - bestDiff = diff;
210   - bestConv = candidate;
211   - }
212   - }
213   -
214   - if (bestConv) {
215   - t.parentTaskId = bestConv.taskId;
216   - // console.log(`[TaskService] 🔗 Inferred parent (master→T): ${t.taskId} → ${bestConv.taskId}`);
217   - }
218   - }
219   -
220   - // 策略3: tpl-train-* / tpl-gen-* / task-* 与 conv-* (master) 的关联
221   - // master 的 conv-xxx 在时间上紧接着触发了 tpl-train-xxx / tpl-gen-xxx
222   - for (const t of allTasks) {
223   - if (t.parentTaskId) continue;
224   -
225   - const isTemplateTask = t.taskId.startsWith('tpl-train-') || t.taskId.startsWith('train-') || t.taskId.startsWith('tpl-gen-');
226   - if (!isTemplateTask) continue;
227   -
228   - const tCreated = new Date(t.createdAt);
229   -
230   - // 找 master 的 conv-xxx,时间上在此刻之前且间隔小于 15 分钟
231   - let bestConv = null;
232   - let bestDiff = 15 * 60 * 1000; // 15分钟窗口
233   -
234   - for (const candidate of allTasks) {
235   - if (!candidate.taskId.startsWith('conv-')) continue;
236   - // 必须有 master 参与
237   - if (!candidate.conversations.some(c => c.agentId === 'master')) continue;
238   -
239   - const cCreated = new Date(candidate.createdAt);
240   - const diff = tCreated - cCreated;
241   -
242   - if (diff > 0 && diff < bestDiff) {
243   - // 额外检查:title 是否有模板名关联
244   - const templateName = t.taskId.replace(/^(tpl-train-|train-|tpl-gen-)/, '').replace(/-[a-z0-9]+$/, '');
245   - if (candidate.title && candidate.title.toLowerCase().includes(templateName.replace(/_/g, ' ').split(' ')[0])) {
246   - bestDiff = diff;
247   - bestConv = candidate;
248   - } else if (!bestConv) {
249   - // 如果没有更好的匹配,用时间最近的
250   - bestDiff = diff;
251   - bestConv = candidate;
252   - }
253   - }
254   - }
255   -
256   - if (bestConv) {
257   - t.parentTaskId = bestConv.taskId;
258   - // console.log(`[TaskService] 🔗 Inferred parent (conv→tpl): ${t.taskId} → ${bestConv.taskId}`);
259   - }
260   - }
261   - }
262   -
263 111 getTaskTimeline(filterTaskId, includeChildren = true) {
264 112 const tasks = this.getAggregatedTasks();
265 113 const taskGroup = tasks.find(t => t.taskId === filterTaskId);
... ... @@ -270,12 +118,9 @@ class TaskService {
270 118 this._collectDescendants(filterTaskId, tasks, taskIds);
271 119 }
272 120  
273   - const relevantConvs = [];
274   - for (const t of tasks) {
275   - if (taskIds.has(t.taskId)) {
276   - relevantConvs.push(...t.conversations);
277   - }
278   - }
  121 + // 会话按「id 里带的 T-xxx」归到任务上(task-T-xxx / callback-T-xxx 都算)
  122 + const relevantConvs = this.getAllConversations()
  123 + .filter((c) => taskIds.has(this._extractTaskId(c.convId)));
279 124  
280 125 let timeline = [];
281 126  
... ...
server/tests/unit/task-dispatch-store.test.js
... ... @@ -92,6 +92,57 @@ describe(&#39;派单存储换成 tasks 表&#39;, () =&gt; {
92 92 expect(row.finished_at).toBeTruthy();
93 93 });
94 94  
  95 + it('派单不被调度器碰 —— 它没有 run 行,出队会把它误收成 cancelled', async () => {
  96 + for (const m of ['../../services/task-scheduler']) delete require_.cache[require_.resolve(m)];
  97 + const scheduler = require_('../../services/task-scheduler');
  98 +
  99 + const d = await post('/api/agent/task/dispatch', {
  100 + from: FROM, to: TO, task: '守卫用例:调度器别碰我', autoExecute: false,
  101 + });
  102 + const taskId = d.body.taskId;
  103 + const row0 = wdb.getDb().prepare('SELECT * FROM tasks WHERE id = ?').get(taskId);
  104 + expect(row0.executor).toBe('agent');
  105 +
  106 + // 把它摆成「最像可出队」的样子:queued + auto + 无依赖。只有 executor 这一维拦得住它。
  107 + wdb.getDb().prepare("UPDATE tasks SET status='queued', execution_mode='auto' WHERE id=?").run(taskId);
  108 +
  109 + expect(scheduler.listQueue(row0.company_id).map((t) => t.id)).not.toContain(taskId);
  110 + scheduler.tick(row0.company_id);
  111 + scheduler.sweep();
  112 +
  113 + // 关键断言:没有被出队、更没有被 abandonRun 收成 cancelled
  114 + const row1 = wdb.getDb().prepare('SELECT * FROM tasks WHERE id = ?').get(taskId);
  115 + expect(row1.status).toBe('queued');
  116 + expect(row1.current_run_id).toBeFalsy();
  117 + });
  118 +
  119 + // 曾经的缺陷:report 的自唤醒回调写死 port 3010,跑一次测试就往线上发一条 CEO chat、
  120 + // 起一个常驻 claude CLI 会话、并留下 system/agents/ceo/_global/callback-T-*。
  121 + it('测试环境下汇报不发对外 HTTP,也不新建 callback 目录', async () => {
  122 + const http = require_('http');
  123 + const orig = http.request;
  124 + const calls = [];
  125 + http.request = function (...args) { calls.push(args[0]); return orig.apply(this, args); };
  126 + const ceoState = path.resolve(process.cwd(), 'system', 'agents', 'ceo', '_global');
  127 + const cbBefore = fs.existsSync(ceoState)
  128 + ? fs.readdirSync(ceoState).filter((n) => n.startsWith('callback-')) : [];
  129 + try {
  130 + const d = await post('/api/agent/task/dispatch', {
  131 + from: FROM, to: TO, task: '守卫用例:回调不外发', autoExecute: false,
  132 + });
  133 + await post('/api/agent/task/report', {
  134 + taskId: d.body.taskId, from: TO, to: FROM, status: 'done', summary: '干完了',
  135 + });
  136 + await new Promise((r) => setTimeout(r, 200)); // 回调走 setImmediate,给它跑的机会
  137 + expect(calls).toEqual([]);
  138 + } finally {
  139 + http.request = orig;
  140 + }
  141 + const cbAfter = fs.existsSync(ceoState)
  142 + ? fs.readdirSync(ceoState).filter((n) => n.startsWith('callback-')) : [];
  143 + expect(cbAfter).toEqual(cbBefore);
  144 + });
  145 +
95 146 it('不再往 system/tasklog/ 写新文件', async () => {
96 147 const dir = path.resolve(process.cwd(), 'system', 'tasklog');
97 148 const before = fs.existsSync(dir) ? fs.readdirSync(dir).length : 0;
... ...