const fs = require('fs'); const path = require('path'); const locator = require('./agent-locator'); const { listCompanyAgents } = require('./company-service'); const SYSTEM_AGENTS_DIR = require('./agent-locator').AGENTS_ROOT; /** * 递归扫描并读取系统中所有 Agent 的会话, * 从 .tasks/ meta 和 conversation 模式中推断父子关系,构建任务树。 */ class TaskService { getAllConversations() { const list = []; if (fs.existsSync(SYSTEM_AGENTS_DIR)) { const agents = fs.readdirSync(SYSTEM_AGENTS_DIR).filter(d => !d.startsWith('.')); for (const agentId of agents) { // 全局 agent 的会话落在保留公司 id `_global` 下(28 §5.1),不再是本体根上 const convDir = path.join(locator.companyDir(agentId), 'conversations'); this._scanConversationsDir(convDir, 'system', agentId, list); } } // 公司 agent 的名册查 company_agents(28 批次 4-19),会话落在公司态目录 // `system/agents/{aid}/{cid}/conversations/` —— 公司侧已经没有 agent 目录可扫了。 for (const row of listCompanyAgents()) { const convDir = path.join(locator.companyDir(row.system_agent_id, row.company_id), 'conversations'); this._scanConversationsDir(convDir, row.company_id, row.role_name, list); } return list; } /** * 从 Conversation 聚合 Task 列表,并通过多种手段推断父子关系: * 1) .tasks/ meta 中的显式 parentTaskId * 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 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, })); } _extractTaskId(convId) { // "task-T-xxx" / "callback-T-xxx" → "T-xxx" if (convId.includes('-T-')) { const match = convId.match(/(T-[a-z0-9]+)/); if (match) return match[1]; } // 其他 (conv-xxx, tpl-train-xxx) 保持原样 return convId; } /** * 某条任务(含子任务)的跨 agent 时间线。 * * 批 5 之后这里是**两个来源合起来**,刻意分清: * · 任务树(有哪些子任务)→ `tasks` 表的 `parent_task_id`,显式、可验证; * · 时间线正文 → agent 的 `conversations/` 目录,那是**一手记录**,没有替代品。 * 删掉的只是「从会话反推任务树」那一半,不是会话本身。 */ getTaskTimeline(filterTaskId, includeChildren = true) { const tasks = this.getAggregatedTasks(); const taskGroup = tasks.find(t => t.taskId === filterTaskId); if (!taskGroup) return []; const taskIds = new Set([filterTaskId]); if (includeChildren) { this._collectDescendants(filterTaskId, tasks, taskIds); } // 会话按「id 里带的 T-xxx」归到任务上(task-T-xxx / callback-T-xxx 都算) const relevantConvs = this.getAllConversations() .filter((c) => taskIds.has(this._extractTaskId(c.convId))); let timeline = []; for (const conv of relevantConvs) { const logPath = path.join(path.dirname(conv.histPath), 'messages.log'); if (fs.existsSync(logPath)) { try { // 限制最大读取 512KB,超出则只读尾部 const MAX_LOG_READ = 512 * 1024; const stat = fs.statSync(logPath); let logContent; if (stat.size > MAX_LOG_READ) { const fd = fs.openSync(logPath, 'r'); const buf = Buffer.alloc(MAX_LOG_READ); fs.readSync(fd, buf, 0, MAX_LOG_READ, stat.size - MAX_LOG_READ); fs.closeSync(fd); logContent = buf.toString('utf-8'); } else { logContent = fs.readFileSync(logPath, 'utf-8'); } const blocks = this._parseMessagesLog(logContent, conv); timeline.push(...blocks); } catch (e) { console.warn('Error parsing messages.log', logPath, e.message); } } else if (fs.existsSync(conv.histPath)) { try { const history = JSON.parse(fs.readFileSync(conv.histPath, 'utf-8')); history.forEach(msg => { if (!msg.content && !msg.tool_calls) return; timeline.push({ key: `${conv.conversationId}-${msg.messageId || Math.random()}`, companyId: conv.companyId, agentId: conv.agentId, conversationId: conv.conversationId, role: msg.role, content: msg.content, timestamp: msg.timestamp || conv.meta.updatedAt, }); }); } catch (e) { console.warn('Error parsing history.json', conv.histPath); } } } timeline.sort((a, b) => new Date(a.timestamp).getTime() - new Date(b.timestamp).getTime()); return timeline; } _collectDescendants(taskId, allTasks, collected) { const task = allTasks.find(t => t.taskId === taskId); if (!task) return; for (const childId of task.children) { if (!collected.has(childId)) { collected.add(childId); this._collectDescendants(childId, allTasks, collected); } } } _parseMessagesLog(logContent, conv) { const blocks = []; const blockRegex = /\[(\d{4}-\d{2}-\d{2}T[\d:.]+Z?)\]\s*\[(USER|AGENT|SYSTEM|TOOL)\]\n/g; const splits = []; let match; while ((match = blockRegex.exec(logContent)) !== null) { splits.push({ timestamp: match[1], role: match[2].toLowerCase(), startIndex: match.index + match[0].length, }); } for (let i = 0; i < splits.length; i++) { const s = splits[i]; const endIndex = (i + 1 < splits.length) ? splits[i + 1].startIndex - splits[i + 1].timestamp.length - 10 : logContent.length; const content = logContent.substring(s.startIndex, endIndex).trim(); if (!content) continue; blocks.push({ key: `${conv.conversationId}-log-${i}`, companyId: conv.companyId, agentId: conv.agentId, conversationId: conv.conversationId, role: s.role === 'agent' ? 'assistant' : s.role, content, timestamp: s.timestamp.endsWith('Z') ? s.timestamp : s.timestamp + 'Z', }); } return blocks; } _scanConversationsDir(convDir, companyId, agentId, list) { if (!fs.existsSync(convDir)) return; const items = fs.readdirSync(convDir, { withFileTypes: true }); for (const item of items) { if (item.isDirectory()) { const metaPath = path.join(convDir, item.name, 'meta.json'); const histPath = path.join(convDir, item.name, 'history.json'); if (fs.existsSync(metaPath)) { let meta = {}; try { meta = JSON.parse(fs.readFileSync(metaPath, 'utf-8')); } catch(e){} list.push({ companyId, agentId, conversationId: item.name, meta, histPath }); } } } } } module.exports = { TaskService };