宜博 / lemonboss

/**
 * 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;