宜博 / pixelphone

/**
 * Master Chat Route — MasterSession SSE 端点
 *
 * POST /api/master/chat  (SSE)
 *   请求: { message, sessionId?, files?: [{fileId, fileName, parsedText}] }
 *   事件: session | progress | agent_output | tool_call | result | error
 */
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 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'));

  let agents = {};
  const agentsPath = path.join(configDir, 'agents.json');
  if (fs.existsSync(agentsPath)) {
    agents = JSON.parse(fs.readFileSync(agentsPath, 'utf-8'));
  }

  const config = {
    ...backends,
    agents,
    agentGroups: {},
    settings: { maxIterations: 10, timeout: 600000, logsDir: '.autodev-logs' },
  };

  if (config.launchers && !config.clients) {
    config.clients = {};
    for (const [name, launcher] of Object.entries(config.launchers)) {
      config.clients[name] = {
        name,
        command: launcher.command,
        acpArgs: launcher.args || [],
        acpProtocol: launcher.protocol || 'acp',
      };
    }
  }

  return config;
}

// ─── Session 池 ───

const sessions = new Map(); // sessionId → { session, lastActive, timer }
const SESSION_IDLE_TIMEOUT = 10 * 60 * 1000;

function getOrCreateSession(sessionId) {
  let entry = sessions.get(sessionId);
  if (entry) {
    entry.lastActive = Date.now();
    if (entry.timer) { clearTimeout(entry.timer); entry.timer = null; }
    return entry.session;
  }

  const config = loadConfig();
  const workDir = require('../config').skillsCodeDir || process.cwd();
  const session = new MasterSession(config, workDir);

  entry = { session, lastActive: Date.now(), timer: null };
  sessions.set(sessionId, entry);
  console.log(`[MasterChat] 新建 session: ${sessionId}`);
  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
 */
router.post('/upload', 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 conversationId = req.body.sessionId || `session-${Date.now()}`;
    const uploadDir = path.join(CONVERSATIONS_DIR, conversationId, 'upload');
    fs.mkdirSync(uploadDir, { recursive: true });

    const safeFileName = file.name.replace(/[^a-zA-Z0-9._\-一-鿿]/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)}`;

    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
 */
router.post('/chat', async (req, res) => {
  const { message, sessionId: reqSessionId, files } = req.body;
  if (!message) return res.status(400).json({ error: 'message is required' });

  const conversationId = reqSessionId || `session-${Date.now()}`;
  const messageId = `msg-${Date.now()}-${randomUUID().substring(0, 8)}`;
  const sendSSE = setupSSE(res);
  const session = getOrCreateSession(conversationId);

  let fullMessage = message;
  if (files && files.length > 0) {
    const fileSection = files.map(f =>
      `--- 📎 文件: ${f.fileName} ---\n${f.parsedText}\n--- 文件结束 ---`
    ).join('\n\n');
    fullMessage = `${message}\n\n[用户上传的文件内容]\n${fileSection}`;
  }

  const logger = new ConversationLogger(CONVERSATIONS_DIR, conversationId, messageId, {});
  const title = message.length > 60 ? message.substring(0, 60) + '...' : message;
  logger.ensureMeta(title);
  logger.logUserMessage(fullMessage);

  sendSSE('session', { sessionId: conversationId });

  try {
    sendSSE('progress', { step: 'initializing', message: '🤖 Master Agent 启动中...' });

    const result = await session.handleMessage(fullMessage, {
      onData: (chunk) => {
        if (chunk && chunk.trim()) {
          sendSSE('agent_output', { agent: 'master', message: chunk, streamType: 'agent_output' });
          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 });

        if (tc.tool === 'call_sub_agent' && tc.params.agent) {
          logger.addStats({ subAgentCount: 1 });
          sendSSE('progress', { step: 'orchestrating', message: `🚀 启动 ${tc.params.agent}...`, agent: tc.params.agent, lifecycle: 'calling' });
        }
      },
      onSubAgentData: (agentName, chunk) => {
        sendSSE('agent_output', { agent: agentName, message: chunk, streamType: 'sub_agent_output' });
        logger.logStreamChunk(agentName, chunk);
      },
      maxToolRounds: 10,
    });

    logger.logResult({ type: result.type, output: result.output, rounds: result.rounds });
    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
 */
router.get('/sessions', (req, res) => {
  const list = [];
  for (const [id, entry] of sessions) {
    list.push({ sessionId: id, lastActive: new Date(entry.lastActive).toISOString(), age: Date.now() - entry.lastActive });
  }
  res.json(list);
});

/**
 * DELETE /api/master/sessions/:id
 */
router.delete('/sessions/:id', async (req, res) => {
  const entry = sessions.get(req.params.id);
  if (!entry) 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;