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; const TASK_META_DIR = path.resolve(__dirname, '../../system/tasklog'); /** * 递归扫描并读取系统中所有 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; } _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 列表,并通过多种手段推断父子关系: * 1) .tasks/ meta 中的显式 parentTaskId * 2) conversationId 模式推断(task-T-xxx 从 tpl-train-* 派出,tpl-train 从 conv-* 派出) * 3) title 中可解析的 task reference([TASK from xxx]) */ 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, })); 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-')) { const match = convId.match(/(T-[a-z0-9]+)/); if (match) return match[1]; } // 其他 (conv-xxx, tpl-train-xxx) 保持原样 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 的父任务 */ _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); if (!taskGroup) return []; const taskIds = new Set([filterTaskId]); if (includeChildren) { this._collectDescendants(filterTaskId, tasks, taskIds); } const relevantConvs = []; for (const t of tasks) { if (taskIds.has(t.taskId)) { relevantConvs.push(...t.conversations); } } 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 };