/** * Master Chat Route — MasterSession SSE 端点 * * POST /api/master/chat (SSE) * 请求: { message, sessionId? } * 事件: session | progress | agent_output | tool_call | result | error * * 维护 MasterSession 池(per sessionId),支持多轮对话。 * 每轮对话写入 conversations/{conversationId}/{messageId}/messages.log */ const express = require('express'); const path = require('path'); const fs = require('fs'); const { randomUUID } = require('crypto'); const fileUpload = require('express-fileupload'); const { parseFileContent } = require('../services/file-parser'); const router = express.Router(); const { MasterSession } = require('../src/core/master-session'); const { ConversationLogger } = require('../services/conversation-logger'); const { requireAuth } = require('../middleware/jwt-auth'); // conversations 根目录 const CONVERSATIONS_DIR = path.resolve(__dirname, '../../conversations'); // ─── 配置加载 ─── function loadConfig() { const configDir = path.resolve(__dirname, '../config'); const backends = JSON.parse(fs.readFileSync(path.join(configDir, 'backends.json'), 'utf-8')); const { loadConfigFromSkills } = require('../src/core/skill-config-loader'); const serverConfig = require('../config'); const skillsDir = serverConfig.skillsCodeDir; return loadConfigFromSkills(skillsDir, backends); } // ─── Session 池 ─── const sessions = new Map(); // sessionId → { session, lastActive, timer, userId } const SESSION_IDLE_TIMEOUT = 10 * 60 * 1000; // 10 分钟空闲超时 function getOrCreateSession(sessionId, userId) { let entry = sessions.get(sessionId); if (entry) { // 校验所有权 if (userId && entry.userId !== String(userId)) { throw new Error('Session belongs to another user'); } entry.lastActive = Date.now(); if (entry.timer) { clearTimeout(entry.timer); entry.timer = null; } return entry.session; } const config = loadConfig(); const serverConfig = require('../config'); const workDir = serverConfig.skillsCodeDir; const session = new MasterSession(config, workDir); entry = { session, lastActive: Date.now(), timer: null, userId: String(userId || '') }; sessions.set(sessionId, entry); console.log(`[MasterChat] 新建 session: ${sessionId} (user: ${userId})`); return session; } function resetIdleTimer(sessionId) { const entry = sessions.get(sessionId); if (!entry) return; if (entry.timer) clearTimeout(entry.timer); entry.timer = setTimeout(async () => { console.log(`[MasterChat] Session idle timeout: ${sessionId}`); try { await entry.session.shutdown(); } catch { } sessions.delete(sessionId); }, SESSION_IDLE_TIMEOUT); } // ─── SSE 设置 ─── function setupSSE(res) { res.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', Connection: 'keep-alive', }); return (event, data) => { res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`); }; } // ─── 文件上传 ─── /** * POST /api/master/upload * * 上传文件,保存到 conversations/{conversationId}/upload/ * 返回: { fileId, fileName, filePath, parsedText, mimeType, truncated } */ router.post('/upload', requireAuth, fileUpload({ limits: { fileSize: 10 * 1024 * 1024 } }), async (req, res) => { try { const file = req.files && req.files.file; if (!file) return res.status(400).json({ error: 'No file uploaded' }); const userId = req.user.id; const conversationId = req.body.sessionId || `session-${Date.now()}`; const uploadDir = path.join(CONVERSATIONS_DIR, `user-${userId}`, conversationId, 'upload'); fs.mkdirSync(uploadDir, { recursive: true }); // 保存文件 const safeFileName = file.name.replace(/[^a-zA-Z0-9._\-\u4e00-\u9fff]/g, '_'); const destPath = path.join(uploadDir, safeFileName); await file.mv(destPath); // 解析内容 const { text, mimeType, truncated } = await parseFileContent(destPath); const fileId = `file-${Date.now()}-${randomUUID().substring(0, 6)}`; console.log(`[MasterChat] 文件上传: ${safeFileName} (${(file.size / 1024).toFixed(1)}KB) → ${conversationId}`); res.json({ fileId, fileName: safeFileName, filePath: destPath, parsedText: text, mimeType, truncated, size: file.size, }); } catch (error) { console.error('[MasterChat] Upload error:', error); res.status(500).json({ error: error.message }); } }); // ─── 路由 ─── /** * POST /api/master/chat * * SSE 流式响应。Master 判断是直接回答还是调用工具/Sub Agent。 * 每轮对话创建 messageId 并写日志到 conversations/{conversationId}/{messageId}/ * 请求体: { message, sessionId?, files?: [{fileId, fileName, parsedText}] } */ router.post('/chat', requireAuth, async (req, res) => { const { message, sessionId: reqSessionId, files } = req.body; if (!message) return res.status(400).json({ error: 'message is required' }); const userId = req.user.id; const conversationId = reqSessionId || `session-${Date.now()}`; const messageId = `msg-${Date.now()}-${randomUUID().substring(0, 8)}`; const sendSSE = setupSSE(res); const session = getOrCreateSession(conversationId, userId); // 拼接文件内容到消息 let fullMessage = message; if (files && files.length > 0) { const fileSection = files.map(f => { return `--- 📎 文件: ${f.fileName} ---\n${f.parsedText}\n--- 文件结束 ---`; }).join('\n\n'); fullMessage = `${message}\n\n[用户上传的文件内容]\n${fileSection}`; } // 创建对话日志 — 按用户隔离存储 const userConvDir = path.join(CONVERSATIONS_DIR, `user-${userId}`); const logger = new ConversationLogger(userConvDir, conversationId, messageId, { userId }); const title = message.length > 60 ? message.substring(0, 60) + '...' : message; logger.ensureMeta(title); logger.logUserMessage(fullMessage); // 通知前端 sessionId(首次对话时前端需要保存) sendSSE('session', { sessionId: conversationId }); // 收集完整输出文本用于日志 let fullOutput = ''; try { // 立即发送进度提示(Claude CLI 冷启动可能需要 30-60 秒) sendSSE('progress', { step: 'initializing', message: '🤖 Master Agent 启动中,首次对话可能需要等待 30-60 秒...', }); const result = await session.handleMessage(fullMessage, { onData: (chunk) => { // Master 自身的流式输出 if (chunk && chunk.trim()) { sendSSE('agent_output', { agent: 'master', message: chunk, streamType: 'agent_output', }); fullOutput += chunk; logger.logStreamChunk('master', chunk); } }, onToolCall: (tc) => { // 工具调用通知 sendSSE('tool_call', { tool: tc.tool, params: tc.params, step: tc.tool, message: `调用工具: ${tc.tool}`, }); logger.logToolCall(tc.tool, JSON.stringify(tc.params || {})); logger.addStats({ toolCallCount: 1 }); // 如果是 call_sub_agent,通知前端创建 Sub Agent tab if (tc.tool === 'call_sub_agent' && tc.params.agent) { logger.addStats({ subAgentCount: 1 }); sendSSE('progress', { step: 'orchestrating', message: `🚀 启动 ${tc.params.agent} 作为 Sub Agent...`, agent: tc.params.agent, lifecycle: 'calling', }); } }, onSubAgentData: (agentName, chunk) => { // Sub Agent 的流式输出 → SSE + 日志 sendSSE('agent_output', { agent: agentName, message: chunk, streamType: 'sub_agent_output', }); logger.logStreamChunk(agentName, chunk); }, maxToolRounds: 10, userId, }); // 记录最终结果 logger.logResult({ type: result.type, output: result.output, rounds: result.rounds, toolCalls: (result.toolCalls || []).length, }); // 记录 cost/duration(如果 result 中包含) if (result.costUsd || result.durationApiMs) { logger.addStats({ ...(result.costUsd ? { costUsd: result.costUsd } : {}), ...(result.durationApiMs ? { durationApiMs: result.durationApiMs } : {}), }); } // 标记对话完成 logger.finalize('completed'); // 发送最终结果 sendSSE('result', { sessionId: conversationId, type: result.type, output: result.output, toolCalls: (result.toolCalls || []).map(tc => ({ tool: tc.tool, params: tc.params, status: tc.result?.status, })), rounds: result.rounds, }); } catch (error) { console.error('[MasterChat] Error:', error); logger.logError(error.message); logger.finalize('failed', error.message); sendSSE('error', { message: error.message }); } finally { resetIdleTimer(conversationId); res.end(); } }); /** * GET /api/master/sessions * * 列出活跃 session */ router.get('/sessions', requireAuth, (req, res) => { const list = []; const userId = String(req.user.id); for (const [id, entry] of sessions) { if (entry.userId !== userId) continue; list.push({ sessionId: id, lastActive: new Date(entry.lastActive).toISOString(), age: Date.now() - entry.lastActive, }); } res.json(list); }); /** * DELETE /api/master/sessions/:id * * 关闭指定 session */ router.delete('/sessions/:id', requireAuth, async (req, res) => { const entry = sessions.get(req.params.id); if (!entry || entry.userId !== String(req.user.id)) { return res.status(404).json({ error: 'Session not found' }); } if (entry.timer) clearTimeout(entry.timer); try { await entry.session.shutdown(); } catch { } sessions.delete(req.params.id); res.json({ success: true }); }); module.exports = router;