feat: add cross conversation messaging

2026-06-12 17:46:37 +08:00
parent 8b2173be8f
commit 04e15c9c89
7 changed files with 1033 additions and 111 deletions
--- a/server.js
+++ b/server.js
@@ -19,14 +19,17 @@ if (fs.existsSync(envPath)) {
 const PORT = parseInt(process.env.PORT) || 8002;
 const CLAUDE_PATH = process.env.CLAUDE_PATH || 'claude';
 const CODEX_PATH = process.env.CODEX_PATH || 'codex';
+const INTERNAL_MCP_TOKEN = process.env.CC_WEB_INTERNAL_MCP_TOKEN || crypto.randomBytes(32).toString('hex');
 const CONFIG_DIR = process.env.CC_WEB_CONFIG_DIR || path.join(__dirname, 'config');
 const SESSIONS_DIR = process.env.CC_WEB_SESSIONS_DIR || path.join(__dirname, 'sessions');
 const PUBLIC_DIR = process.env.CC_WEB_PUBLIC_DIR || path.join(__dirname, 'public');
 const LOGS_DIR = process.env.CC_WEB_LOGS_DIR || path.join(__dirname, 'logs');
+const CCWEB_MCP_SERVER_PATH = path.join(__dirname, 'lib', 'ccweb-mcp-server.js');
 const ATTACHMENTS_DIR = path.join(SESSIONS_DIR, '_attachments');
 const ATTACHMENT_TTL_MS = 7 * 24 * 60 * 60 * 1000;
 const MAX_ATTACHMENT_SIZE = 10 * 1024 * 1024;
 const MAX_MESSAGE_ATTACHMENTS = 4;
+const CROSS_CONVERSATION_MAX_HOPS = 1;
 const FILE_BROWSER_MAX_LIST_ENTRIES = 400;
 const FILE_BROWSER_MAX_PREVIEW_BYTES = 200 * 1024;
 const IMAGE_MIME_TYPES = new Set(['image/png', 'image/jpeg', 'image/webp', 'image/gif']);
@@ -1007,6 +1010,43 @@ function jsonResponse(res, statusCode, payload) {
  res.end(JSON.stringify(payload));
 }

+function readJsonBody(req, maxBytes = 1024 * 1024) {
+  return new Promise((resolve, reject) => {
+    const chunks = [];
+    let total = 0;
+    let finished = false;
+
+    req.on('data', (chunk) => {
+      if (finished) return;
+      total += chunk.length;
+      if (total > maxBytes) {
+        finished = true;
+        reject(new Error('请求体过大'));
+        req.destroy();
+        return;
+      }
+      chunks.push(chunk);
+    });
+
+    req.on('end', () => {
+      if (finished) return;
+      finished = true;
+      try {
+        const text = Buffer.concat(chunks).toString('utf8').trim();
+        resolve(text ? JSON.parse(text) : {});
+      } catch {
+        reject(new Error('请求体不是有效 JSON'));
+      }
+    });
+
+    req.on('error', (err) => {
+      if (finished) return;
+      finished = true;
+      reject(err);
+    });
+  });
+}
+
 function normalizeRelativeBrowserPath(input) {
  return String(input || '')
    .replace(/\\/g, '/')
@@ -1585,6 +1625,196 @@ function sendSessionList(ws) {
  }
 }

+function broadcastSessionList() {
+  if (!wss) return;
+  for (const client of wss.clients) {
+    if (client.readyState === 1) sendSessionList(client);
+  }
+}
+
+function findViewingSessionWs(sessionId) {
+  const normalizedId = sanitizeId(sessionId || '');
+  if (!normalizedId) return null;
+  for (const [client, viewedSessionId] of wsSessionMap.entries()) {
+    if (viewedSessionId === normalizedId && client?.readyState === 1) return client;
+  }
+  return null;
+}
+
+function getInternalMcpRequestToken(req) {
+  return String(req.headers['x-cc-web-mcp-token'] || '').trim() || extractBearerToken(req);
+}
+
+function mcpToolError(code, message, extra = {}) {
+  return { ok: false, code, message, ...extra };
+}
+
+function clampMcpLimit(value) {
+  const parsed = Number.parseInt(String(value || ''), 10);
+  if (!Number.isFinite(parsed)) return 50;
+  return Math.max(1, Math.min(100, parsed));
+}
+
+function listConversationSummaries(args = {}, sourceSessionId = '') {
+  const agentFilter = VALID_AGENTS.has(args.agent) ? args.agent : '';
+  const statusFilter = ['running', 'idle'].includes(args.status) ? args.status : 'all';
+  const limit = clampMcpLimit(args.limit);
+  const conversations = [];
+
+  try {
+    const files = fs.readdirSync(SESSIONS_DIR).filter((name) => name.endsWith('.json'));
+    for (const file of files) {
+      try {
+        const session = normalizeSession(JSON.parse(fs.readFileSync(path.join(SESSIONS_DIR, file), 'utf8')));
+        const agent = getSessionAgent(session);
+        const running = activeProcesses.has(session.id);
+        const status = running ? 'running' : 'idle';
+        if (agentFilter && agent !== agentFilter) continue;
+        if (statusFilter !== 'all' && status !== statusFilter) continue;
+        const cwd = session.cwd || '';
+        conversations.push({
+          id: session.id,
+          title: session.title || 'Untitled',
+          agent,
+          status,
+          updatedAt: session.updated || session.created || null,
+          cwd,
+          projectName: cwd ? path.basename(cwd.replace(/[\\/]+$/, '')) : '',
+          isCurrent: session.id === sourceSessionId,
+        });
+      } catch {}
+    }
+  } catch {}
+
+  conversations.sort((a, b) => new Date(b.updatedAt || 0) - new Date(a.updatedAt || 0));
+  return {
+    ok: true,
+    currentConversationId: sourceSessionId || null,
+    conversations: conversations.slice(0, limit),
+  };
+}
+
+function buildCrossConversationRuntimeText(sourceSession, content) {
+  const sourceTitle = sourceSession?.title || 'Untitled';
+  const sourceId = sourceSession?.id || '';
+  return `来自「${sourceTitle}」对话（ID: ${sourceId}）的消息：\n\n${content}`;
+}
+
+function sendCrossConversationMessage(args = {}, sourceSessionId = '', sourceHopCount = 0) {
+  const sourceId = sanitizeId(sourceSessionId || '');
+  const targetId = sanitizeId(args.targetConversationId || args.targetSessionId || args.conversationId || '');
+  const content = typeof args.content === 'string' ? args.content.trim() : '';
+
+  if (!sourceId) {
+    return mcpToolError('missing_source_conversation', '缺少来源对话 ID。');
+  }
+  if (!targetId) {
+    return mcpToolError('missing_target_conversation', '缺少目标对话 ID。');
+  }
+  if (!content) {
+    return mcpToolError('empty_content', '消息内容不能为空。');
+  }
+
+  const sourceSession = loadSession(sourceId);
+  if (!sourceSession) {
+    return mcpToolError('source_not_found', '来源对话不存在。', { sourceConversationId: sourceId });
+  }
+
+  const targetSession = loadSession(targetId);
+  if (!targetSession) {
+    return mcpToolError('target_not_found', '目标对话不存在。', { targetConversationId: targetId });
+  }
+
+  if (sourceId === targetId) {
+    return mcpToolError('same_conversation', '不能把消息发送给当前同一个对话。', { targetConversationId: targetId });
+  }
+
+  const normalizedHopCount = Math.max(0, Number.parseInt(String(sourceHopCount || 0), 10) || 0);
+  if (normalizedHopCount >= CROSS_CONVERSATION_MAX_HOPS) {
+    return mcpToolError('hop_limit_exceeded', '跨对话消息已达到转发跳数上限，已拒绝继续转发。', {
+      sourceConversationId: sourceId,
+      targetConversationId: targetId,
+      maxHops: CROSS_CONVERSATION_MAX_HOPS,
+    });
+  }
+
+  if (activeProcesses.has(targetId)) {
+    return mcpToolError('target_running', '目标对话正在处理中，请稍后再发送。', { targetConversationId: targetId });
+  }
+
+  const now = new Date().toISOString();
+  const messageId = crypto.randomUUID();
+  const crossConversation = {
+    messageId,
+    sourceSessionId: sourceSession.id,
+    sourceTitle: sourceSession.title || 'Untitled',
+    sentAt: now,
+    hopCount: normalizedHopCount + 1,
+  };
+
+  const targetWs = findViewingSessionWs(targetId);
+  const result = handleMessage(targetWs, {
+    text: content,
+    sessionId: targetSession.id,
+    mode: targetSession.permissionMode || 'yolo',
+    agent: getSessionAgent(targetSession),
+  }, {
+    crossConversation,
+    emitUserMessage: true,
+    runtimeText: buildCrossConversationRuntimeText(sourceSession, content),
+    mcpContext: { hopCount: crossConversation.hopCount },
+  });
+
+  if (!result?.ok) {
+    return mcpToolError(result?.code || 'send_failed', result?.message || '跨对话消息发送失败。', {
+      sourceConversationId: sourceId,
+      targetConversationId: targetId,
+    });
+  }
+
+  broadcastSessionList();
+  return {
+    ok: true,
+    messageId,
+    deliveryStatus: 'delivered',
+    sourceConversationId: sourceId,
+    targetConversationId: targetId,
+    targetTitle: targetSession.title || 'Untitled',
+  };
+}
+
+function callInternalMcpTool(tool, args, sourceSessionId, sourceHopCount) {
+  switch (tool) {
+    case 'ccweb_list_conversations':
+      return listConversationSummaries(args, sanitizeId(sourceSessionId || ''));
+    case 'ccweb_send_message':
+      return sendCrossConversationMessage(args, sourceSessionId, sourceHopCount);
+    default:
+      return mcpToolError('unknown_tool', `未知 MCP 工具: ${tool}`);
+  }
+}
+
+async function handleInternalMcpApi(req, res) {
+  const token = getInternalMcpRequestToken(req);
+  if (!token || token !== INTERNAL_MCP_TOKEN) {
+    return jsonResponse(res, 401, mcpToolError('unauthorized', 'MCP 内部接口未授权。'));
+  }
+
+  let payload;
+  try {
+    payload = await readJsonBody(req);
+  } catch (err) {
+    return jsonResponse(res, 400, mcpToolError('bad_request', err.message || '请求体无效。'));
+  }
+
+  const tool = String(payload.tool || '');
+  const args = payload.args && typeof payload.args === 'object' ? payload.args : {};
+  const sourceSessionId = sanitizeId(payload.sourceSessionId || '');
+  const sourceHopCount = Number.parseInt(String(payload.sourceHopCount || 0), 10) || 0;
+  const result = callInternalMcpTool(tool, args, sourceSessionId, sourceHopCount);
+  return jsonResponse(res, result.ok ? 200 : 400, result);
+}
+
 // === File Tailer ===
 // Tails a file and calls onLine for each new complete line.
 class FileTailer {
@@ -2016,6 +2246,10 @@ function recoverProcesses() {
 const server = http.createServer((req, res) => {
  const url = new URL(req.url, `http://${req.headers.host}`);

+  if (req.method === 'POST' && url.pathname === '/api/internal/mcp') {
+    return handleInternalMcpApi(req, res);
+  }
+
  if (req.method === 'POST' && url.pathname === '/api/attachments') {
    const token = extractBearerToken(req);
    if (!token || !activeTokens.has(token)) {
@@ -2160,7 +2394,26 @@ const server = http.createServer((req, res) => {
 });

 // === WebSocket Server ===
-const wss = new WebSocketServer({ server });
+const wss = new WebSocketServer({ noServer: true });
+
+server.on('upgrade', (req, socket, head) => {
+  let pathname = '';
+  try {
+    pathname = new URL(req.url || '/', 'http://localhost').pathname;
+  } catch {
+    pathname = '';
+  }
+
+  if (pathname !== '/ws') {
+    socket.write('HTTP/1.1 404 Not Found\r\nConnection: close\r\n\r\n');
+    socket.destroy();
+    return;
+  }
+
+  wss.handleUpgrade(req, socket, head, (ws) => {
+    wss.emit('connection', ws, req);
+  });
+});

 wss.on('connection', (ws, req) => {
  ws._req = req;
@@ -3042,14 +3295,22 @@ function handleAbort(ws) {
 function handleMessage(ws, msg, options = {}) {
  const { text, sessionId, mode } = msg;
  const { hideInHistory = false } = options;
+  const fail = (code, message) => {
+    wsSend(ws, { type: 'error', code, message });
+    return { ok: false, code, message };
+  };
  const textValue = typeof text === 'string' ? text : '';
+  const runtimeTextValue = typeof options.runtimeText === 'string' ? options.runtimeText : textValue;
  const attachments = Array.isArray(msg.attachments) ? msg.attachments.slice(0, MAX_MESSAGE_ATTACHMENTS) : [];
  const normalizedText = textValue.trim();
+  const normalizedRuntimeText = runtimeTextValue.trim();
  const resolvedAttachments = resolveMessageAttachments(attachments);
  if (attachments.length > 0 && resolvedAttachments.length === 0) {
-    return wsSend(ws, { type: 'error', message: '图片附件已过期或不可用，请重新上传后再发送。' });
+    return fail('attachment_unavailable', '图片附件已过期或不可用，请重新上传后再发送。');
+  }
+  if (!normalizedText && resolvedAttachments.length === 0) {
+    return fail('empty_message', '消息内容不能为空。');
  }
-  if (!normalizedText && resolvedAttachments.length === 0) return;

  const savedAttachments = resolvedAttachments.map((attachment) => ({
    id: attachment.id,
@@ -3063,7 +3324,7 @@ function handleMessage(ws, msg, options = {}) {
  }));

  if (sessionId && activeProcesses.has(sessionId)) {
-    return wsSend(ws, { type: 'error', message: '正在处理中，请先点击停止按钮。' });
+    return fail('session_running', '正在处理中，请先点击停止按钮。');
  }

  const derivedTitle = normalizedText
@@ -3095,38 +3356,45 @@ function handleMessage(ws, msg, options = {}) {
  normalizeSession(session);

  if (normalizedText.startsWith('/') && resolvedAttachments.length > 0) {
-    return wsSend(ws, { type: 'error', message: '命令消息暂不支持同时附带图片。请先发送图片说明，再单独使用 /model 或 /mode。' });
+    return fail('command_attachment_unsupported', '命令消息暂不支持同时附带图片。请先发送图片说明，再单独使用 /model 或 /mode。');
  }

  if (mode && ['default', 'plan', 'yolo'].includes(mode)) {
    session.permissionMode = mode;
  }

-  if (!hideInHistory && normalizedText !== '/compact' && getRuntimeSessionId(session)) {
-    pendingCompactRetries.set(session.id, { text: normalizedText, mode: session.permissionMode || 'yolo', reason: 'normal' });
+  if (!hideInHistory && normalizedRuntimeText !== '/compact' && getRuntimeSessionId(session)) {
+    pendingCompactRetries.set(session.id, { text: normalizedRuntimeText, mode: session.permissionMode || 'yolo', reason: 'normal' });
  }

  if (session.title === 'New Chat' || session.title === 'Untitled') {
    session.title = derivedTitle;
  }

+  let persistedUserMessage = null;
  if (!hideInHistory) {
-    session.messages.push({
+    persistedUserMessage = {
      role: 'user',
      content: textValue,
      attachments: savedAttachments,
      timestamp: new Date().toISOString(),
-    });
+    };
+    if (options.crossConversation) {
+      persistedUserMessage.crossConversation = options.crossConversation;
+    }
+    session.messages.push(persistedUserMessage);
  }
  session.updated = new Date().toISOString();
  saveSession(session);

  const currentSessionId = session.id;

-  for (const [, entry] of activeProcesses) {
-    if (entry.ws === ws) entry.ws = null;
+  if (ws) {
+    for (const [, entry] of activeProcesses) {
+      if (entry.ws === ws) entry.ws = null;
+    }
+    wsSessionMap.set(ws, currentSessionId);
  }
-  wsSessionMap.set(ws, currentSessionId);

  if (!sessionId) {
    wsSend(ws, {
@@ -3146,13 +3414,20 @@ function handleMessage(ws, msg, options = {}) {
      isRunning: false,
    });
  }
+  if (ws && options.emitUserMessage && persistedUserMessage) {
+    wsSend(ws, { type: 'session_message', sessionId: currentSessionId, message: persistedUserMessage });
+  }
  sendSessionList(ws);

+  const runtimeOptions = {
+    attachments: resolvedAttachments,
+    mcpContext: options.mcpContext || {},
+  };
  const spawnSpec = isClaudeSession(session)
-    ? buildClaudeSpawnSpec(session, { attachments: resolvedAttachments })
-    : buildCodexSpawnSpec(session, { attachments: resolvedAttachments });
+    ? buildClaudeSpawnSpec(session, runtimeOptions)
+    : buildCodexSpawnSpec(session, runtimeOptions);
  if (spawnSpec?.error) {
-    return wsSend(ws, { type: 'error', message: spawnSpec.error });
+    return fail('runtime_config_error', spawnSpec.error);
  }

  // === Detached process with file-based I/O ===
@@ -3165,7 +3440,7 @@ function handleMessage(ws, msg, options = {}) {

  if (isClaudeSession(session) && resolvedAttachments.length > 0) {
    const content = [];
-    if (textValue) content.push({ type: 'text', text: textValue });
+    if (runtimeTextValue) content.push({ type: 'text', text: runtimeTextValue });
    for (const attachment of resolvedAttachments) {
      const data = fs.readFileSync(attachment.path).toString('base64');
      content.push({
@@ -3185,7 +3460,7 @@ function handleMessage(ws, msg, options = {}) {
      },
    })}\n`);
  } else {
-    fs.writeFileSync(inputPath, textValue);
+    fs.writeFileSync(inputPath, runtimeTextValue);
  }

  const inputFd = fs.openSync(inputPath, 'r');
@@ -3208,7 +3483,7 @@ function handleMessage(ws, msg, options = {}) {
    cleanRunDir(currentSessionId);
    plog('ERROR', 'process_spawn_fail', { sessionId: currentSessionId.slice(0, 8), error: err.message });
    const agent = getSessionAgent(session);
-    return wsSend(ws, { type: 'error', message: formatRuntimeError(agent, err.message, { exitCode: null, signal: null }) });
+    return fail('process_spawn_failed', formatRuntimeError(agent, err.message, { exitCode: null, signal: null }));
  }

  fs.closeSync(inputFd);
@@ -3225,7 +3500,7 @@ function handleMessage(ws, msg, options = {}) {
    mode: spawnSpec.mode,
    model: session.model || 'default',
    resume: spawnSpec.resume,
-    args: spawnSpec.args.join(' '),
+    args: redactSpawnArgs(spawnSpec.args.join(' ')),
  });

  // Fast exit detection (while Node.js is running)
@@ -3265,6 +3540,8 @@ function handleMessage(ws, msg, options = {}) {
    } catch {}
  });
  entry.tailer.start();
+
+  return { ok: true, sessionId: currentSessionId, pid: proc.pid };
 }

 function truncateObj(obj, maxLen) {
@@ -3296,6 +3573,12 @@ function sanitizeToolInput(toolName, input) {
  return truncateObj(parsed, 500);
 }

+function redactSpawnArgs(argsText) {
+  return String(argsText || '')
+    .replace(/CC_WEB_MCP_TOKEN[^\s,\]}]*/g, 'CC_WEB_MCP_TOKEN=****')
+    .replace(/mcp_servers\.ccweb\.env=\{[^}]*\}/g, 'mcp_servers.ccweb.env={****}');
+}
+
 const {
  buildClaudeSpawnSpec,
  buildCodexSpawnSpec,
@@ -3312,6 +3595,10 @@ const {
  getDefaultCodexModel,
  loadCodexConfig,
  prepareCodexCustomRuntime,
+  ccwebMcpServerPath: CCWEB_MCP_SERVER_PATH,
+  internalMcpUrl: `http://127.0.0.1:${PORT}/api/internal/mcp`,
+  internalMcpToken: INTERNAL_MCP_TOKEN,
+  nodePath: process.execPath,
  wsSend,
  truncateObj,
  sanitizeToolInput,