diff --git a/server/routes/tasks.js b/server/routes/tasks.js index c89ca37..2918858 100644 --- a/server/routes/tasks.js +++ b/server/routes/tasks.js @@ -30,7 +30,9 @@ router.get('/', (req, res) => { children: t.children, isRoot: t.isRoot, agents: t.agents, - conversationCount: t.conversations.length, + // 批 5：会话数由 TaskService 直接给（原来这里自己取 t.conversations.length， + // 而 conversations 是会话反推那套的产物，已随之删除） + conversationCount: t.conversationCount, })); _taskListCache = taskList; diff --git a/server/services/task-scheduler.js b/server/services/task-scheduler.js index 8714187..8095088 100644 --- a/server/services/task-scheduler.js +++ b/server/services/task-scheduler.js @@ -111,12 +111,18 @@ function dependenciesSatisfied(task) { * * @returns {object[]} 该公司全部 queued task（含 manual 的：它们在界面上同样占位次） */ +// 批 5：只认 executor='graph'。agent 派单（executor='agent'）虽然也进 tasks 表， +// 但它的执行载体是「直接起一次 agent 对话」，由 /task/dispatch 自己触发 —— +// 进表只为了可见、可查、可串成派单树，**不该被调度器出队**。 +// 漏了这一条的后果是静默的：调度器会拿一条根本没有 run 行的 task 去 syncRunNodesToDef， +// 5a 判空 → abandonRun → 把一条正在被 agent 执行的派单收成 cancelled。 function listQueue(companyId) { return wdb() .getDb() .prepare(` SELECT * FROM tasks WHERE company_id = @companyId AND status = 'queued' AND archived_at IS NULL + AND executor = 'graph' ORDER BY priority, seq, created_at `) .all({ companyId }); @@ -156,7 +162,10 @@ function deriveQueueBlocks(companyId) { const requeued = []; for (let pass = 0; ; pass++) { - const rows = listCompanyTasks(companyId).filter((t) => t.status === 'queued' || t.status === 'blocked'); + // 同上：依赖推导也只管图任务。派单任务没有 depends_on 语义（它的先后靠 parent_task_id 表达）， + // 放进来只会让它被判成「依赖不可能满足」而无辜转 blocked。 + const rows = listCompanyTasks(companyId) + .filter((t) => t.executor === 'graph' && (t.status === 'queued' || t.status === 'blocked')); let changed = 0; for (const t of rows) { const dead = deadDependency(t); @@ -373,6 +382,7 @@ function sweep() { .prepare(` SELECT DISTINCT company_id FROM tasks WHERE archived_at IS NULL + AND executor = 'graph' AND (status = 'queued' OR (status = 'blocked' AND blocked_reason LIKE @prefix)) `) diff --git a/server/services/task-service.js b/server/services/task-service.js index 5bed5b7..1914119 100644 --- a/server/services/task-service.js +++ b/server/services/task-service.js @@ -4,7 +4,6 @@ const locator = require('./agent-locator'); const { listCompanyAgents } = require('./company-service'); const SYSTEM_AGENTS_DIR = require('./agent-locator').AGENTS_ROOT; -const TASK_META_DIR = path.resolve(__dirname, '../../system/tasklog'); /** * 递归扫描并读取系统中所有 Agent 的会话， @@ -34,18 +33,6 @@ class TaskService { return list; } - _loadTaskMetas() { - const metas = new Map(); - if (!fs.existsSync(TASK_META_DIR)) return metas; - const files = fs.readdirSync(TASK_META_DIR).filter(f => f.endsWith('.json')); - for (const f of files) { - try { - const tm = JSON.parse(fs.readFileSync(path.join(TASK_META_DIR, f), 'utf-8')); - if (tm.taskId) metas.set(tm.taskId, tm); - } catch(e) {} - } - return metas; - } /** * 从 Conversation 聚合 Task 列表，并通过多种手段推断父子关系： @@ -53,67 +40,55 @@ class TaskService { * 2) conversationId 模式推断（task-T-xxx 从 tpl-train-* 派出，tpl-train 从 conv-* 派出） * 3) title 中可解析的 task reference（[TASK from xxx]） */ + /** + * 任务列表 —— 批 5 起**读 tasks 表**，不再从会话记录反推。 + * + * 原来这里是 122 行的启发式：扫全部 agent 的 conversations，按会话 id 里的 `T-xxx` 归组， + * 再用三条策略（meta 的 parentTaskId / from agent 的活跃会话时间包含 / master conv 的 + * 标题相似）**猜**父子关系。猜出来的树没法验证对错，而且每次请求都要遍历所有 agent 的 + * 会话目录（route 那 10 秒缓存就是为它加的）。 + * + * 现在父子关系是 `tasks.parent_task_id` —— 派单时就显式写下来了，不用猜。 + * + * **响应字段一个没少**（TaskTimeline.vue 按名取值）。两处口径变化写在下面。 + */ getAggregatedTasks() { - const convs = this.getAllConversations(); - const taskMetas = this._loadTaskMetas(); - const taskMap = new Map(); - - // 第一遍：建立所有 task group - for (const c of convs) { - let taskId = this._extractTaskId(c.conversationId); - - if (!taskMap.has(taskId)) { - const meta = taskMetas.get(taskId); - taskMap.set(taskId, { - taskId, - title: c.meta.title || taskId, - createdAt: c.meta.createdAt || new Date().toISOString(), - updatedAt: c.meta.updatedAt || c.meta.createdAt, - status: meta?.status || c.meta.status || 'unknown', - from: meta?.from || null, - to: meta?.to || null, - parentTaskId: meta?.parentTaskId || null, - agents: new Set(), - conversations: [], - children: [], - }); - } - - const t = taskMap.get(taskId); - t.agents.add(`${c.companyId}:${c.agentId}`); - t.conversations.push(c); - - if (c.meta.createdAt && new Date(c.meta.createdAt) < new Date(t.createdAt)) t.createdAt = c.meta.createdAt; - if (c.meta.updatedAt && new Date(c.meta.updatedAt) > new Date(t.updatedAt)) t.updatedAt = c.meta.updatedAt; - } - - // 第二遍：推断父子关系 - this._inferParentChild(taskMap, taskMetas); - - // 第三遍：构建 children 列表 - for (const t of taskMap.values()) { - if (t.parentTaskId && taskMap.has(t.parentTaskId)) { - const parent = taskMap.get(t.parentTaskId); - if (!parent.children.includes(t.taskId)) { - parent.children.push(t.taskId); - } - } - } - - const tasks = Array.from(taskMap.values()).map(t => ({ - ...t, - agents: Array.from(t.agents), - isRoot: !t.parentTaskId || !taskMap.has(t.parentTaskId), - conversationCount: t.conversations.length, + const wdb = require('./work-graph-db'); + const rows = wdb.getDb().prepare(` + SELECT * FROM tasks WHERE archived_at IS NULL ORDER BY created_at DESC + `).all(); + + // 表里的取值域是 tasks 的，而这个端点历史上吐的是 meta 的（dispatched/done）。 + // 前端按字符串比对状态，所以这里翻译回去 —— 换掉存储不该让界面跟着改。 + const TO_META = { running: 'dispatched', completed: 'done', queued: 'dispatched', draft: 'dispatched' }; + const childrenOf = new Map(); + for (const r of rows) { + if (!r.parent_task_id) continue; + if (!childrenOf.has(r.parent_task_id)) childrenOf.set(r.parent_task_id, []); + childrenOf.get(r.parent_task_id).push(r.id); + } + + return rows.map((r) => ({ + taskId: r.id, + title: r.title, + createdAt: r.created_at, + updatedAt: r.updated_at, + status: TO_META[r.status] || r.status, + from: r.created_by, + to: r.assignee_agent_id, + parentTaskId: r.parent_task_id, + children: childrenOf.get(r.id) || [], + isRoot: !r.parent_task_id, + // agents：原来是「会话里出现过的 agent 集合」，现在是「派给谁 + 谁派的」。 + // 对派单任务这两者本来就是同一件事；图任务没有派单对象，只有 created_by。 + agents: [r.created_by, r.assignee_agent_id].filter(Boolean).filter((v, i, a) => a.indexOf(v) === i), + // conversationCount：原来是扫出来的会话文件数，现在是「有没有起过执行会话」（0/1）。 + // 为一个列表字段去遍历所有 agent 的会话目录，正是这次要去掉的那笔开销； + // 真要看会话内容走 /api/tasks/:taskId/timeline，那条仍读一手记录。 + conversationCount: r.execution_conv_id ? 1 : 0, })); - - tasks.sort((a, b) => new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime()); - return tasks; } - /** - * 从 conversationId 提取逻辑 taskId - */ _extractTaskId(convId) { // "task-T-xxx" / "callback-T-xxx" → "T-xxx" if (convId.includes('-T-')) { @@ -124,142 +99,15 @@ class TaskService { return convId; } + /** - * 多策略推断父子关系： - * - * 策略1: .tasks/ meta 中的 parentTaskId（直接有的情况） - * 策略2: 通过 "from" agent 的 活跃会话 推断 - * - T-xxx 的 meta.from = "innovator" - * - innovator 同时有一个 tpl-train-* 会话（时间上包含 T-xxx） - * → tpl-train-* 就是 T-xxx 的父任务 - * 策略3: 通过 master conv 触发 tpl-train 推断 - * - conv-xxx (master) 的时间 < tpl-train-xxx 的时间 且 title 包含相同模板名 - * → conv-xxx 是 tpl-train-xxx 的父任务 + * 某条任务（含子任务）的跨 agent 时间线。 + * + * 批 5 之后这里是**两个来源合起来**，刻意分清： + * · 任务树（有哪些子任务）→ `tasks` 表的 `parent_task_id`，显式、可验证； + * · 时间线正文 → agent 的 `conversations/` 目录，那是**一手记录**，没有替代品。 + * 删掉的只是「从会话反推任务树」那一半，不是会话本身。 */ - _inferParentChild(taskMap, taskMetas) { - const allTasks = Array.from(taskMap.values()); - - // 策略1: 已有 parentTaskId (来自 .tasks/ meta)，已在构建时设置 - - // 策略2: T-xxx 任务，推断父任务为 from agent 的活跃 tpl-train/task 会话 - for (const t of allTasks) { - if (t.parentTaskId) continue; - if (!t.taskId.startsWith('T-')) continue; - - const meta = taskMetas.get(t.taskId); - if (!meta) continue; - - const fromAgent = meta.from; - const taskCreatedAt = new Date(meta.createdAt); - - // 找 fromAgent 的其他会话，时间上在这个 task 之前开始且之后结束/仍活跃 - let bestParent = null; - let bestTimeDiff = Infinity; - - for (const candidate of allTasks) { - if (candidate.taskId === t.taskId) continue; - if (candidate.taskId.startsWith('T-') && !taskMetas.has(candidate.taskId)) continue; - - const hasAgent = candidate.conversations.some(c => c.agentId === fromAgent); - if (!hasAgent) continue; - - const candidateCreated = new Date(candidate.createdAt); - const candidateUpdated = new Date(candidate.updatedAt); - - if (candidateCreated < taskCreatedAt && candidateUpdated >= taskCreatedAt) { - const diff = taskCreatedAt - candidateCreated; - if (diff < bestTimeDiff) { - bestTimeDiff = diff; - bestParent = candidate; - } - } - } - - if (bestParent) { - t.parentTaskId = bestParent.taskId; - // console.log(`[TaskService] 🔗 Inferred parent: ${t.taskId} → ${bestParent.taskId}`); - } - } - - // 策略4: T-xxx from master，但 master 的 conv-* 已经结束的情况 - // 这种情况发生在：master 原始对话结束后，通过 callback 链路继续派发新任务 - // 例如 conv-xxx → tpl-train → T-a (callback → master → 新 dispatch T-b) - // 此时 T-b 应该关联到 conv-xxx（同一个用户请求） - for (const t of allTasks) { - if (t.parentTaskId) continue; - if (!t.taskId.startsWith('T-')) continue; - - const meta = taskMetas.get(t.taskId); - if (!meta || meta.from !== 'master') continue; - - const taskCreatedAt = new Date(meta.createdAt); - - // 找最近的 master conv-* 会话，窗口放宽到 30 分钟 - let bestConv = null; - let bestDiff = 30 * 60 * 1000; - - for (const candidate of allTasks) { - if (!candidate.taskId.startsWith('conv-')) continue; - if (!candidate.conversations.some(c => c.agentId === 'master')) continue; - - const cCreated = new Date(candidate.createdAt); - const diff = taskCreatedAt - cCreated; - - if (diff > 0 && diff < bestDiff) { - bestDiff = diff; - bestConv = candidate; - } - } - - if (bestConv) { - t.parentTaskId = bestConv.taskId; - // console.log(`[TaskService] 🔗 Inferred parent (master→T): ${t.taskId} → ${bestConv.taskId}`); - } - } - - // 策略3: tpl-train-* / tpl-gen-* / task-* 与 conv-* (master) 的关联 - // master 的 conv-xxx 在时间上紧接着触发了 tpl-train-xxx / tpl-gen-xxx - for (const t of allTasks) { - if (t.parentTaskId) continue; - - const isTemplateTask = t.taskId.startsWith('tpl-train-') || t.taskId.startsWith('train-') || t.taskId.startsWith('tpl-gen-'); - if (!isTemplateTask) continue; - - const tCreated = new Date(t.createdAt); - - // 找 master 的 conv-xxx，时间上在此刻之前且间隔小于 15 分钟 - let bestConv = null; - let bestDiff = 15 * 60 * 1000; // 15分钟窗口 - - for (const candidate of allTasks) { - if (!candidate.taskId.startsWith('conv-')) continue; - // 必须有 master 参与 - if (!candidate.conversations.some(c => c.agentId === 'master')) continue; - - const cCreated = new Date(candidate.createdAt); - const diff = tCreated - cCreated; - - if (diff > 0 && diff < bestDiff) { - // 额外检查：title 是否有模板名关联 - const templateName = t.taskId.replace(/^(tpl-train-|train-|tpl-gen-)/, '').replace(/-[a-z0-9]+$/, ''); - if (candidate.title && candidate.title.toLowerCase().includes(templateName.replace(/_/g, ' ').split(' ')[0])) { - bestDiff = diff; - bestConv = candidate; - } else if (!bestConv) { - // 如果没有更好的匹配，用时间最近的 - bestDiff = diff; - bestConv = candidate; - } - } - } - - if (bestConv) { - t.parentTaskId = bestConv.taskId; - // console.log(`[TaskService] 🔗 Inferred parent (conv→tpl): ${t.taskId} → ${bestConv.taskId}`); - } - } - } - getTaskTimeline(filterTaskId, includeChildren = true) { const tasks = this.getAggregatedTasks(); const taskGroup = tasks.find(t => t.taskId === filterTaskId); @@ -270,12 +118,9 @@ class TaskService { this._collectDescendants(filterTaskId, tasks, taskIds); } - const relevantConvs = []; - for (const t of tasks) { - if (taskIds.has(t.taskId)) { - relevantConvs.push(...t.conversations); - } - } + // 会话按「id 里带的 T-xxx」归到任务上（task-T-xxx / callback-T-xxx 都算） + const relevantConvs = this.getAllConversations() + .filter((c) => taskIds.has(this._extractTaskId(c.convId))); let timeline = []; diff --git a/server/tests/unit/task-dispatch-store.test.js b/server/tests/unit/task-dispatch-store.test.js index f319d5d..8929b83 100644 --- a/server/tests/unit/task-dispatch-store.test.js +++ b/server/tests/unit/task-dispatch-store.test.js @@ -92,6 +92,57 @@ describe('派单存储换成 tasks 表', () => { 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;