memind/direct-chat-service.mjs

import crypto from 'node:crypto';

export const DIRECT_CHAT_SESSION_PREFIX = 'h5direct_';

function envFlag(value, fallback = false) {
  const raw = String(value ?? '').trim().toLowerCase();
  if (!raw) return fallback;
  return ['1', 'true', 'yes', 'on'].includes(raw);
}

export function isDirectChatSessionId(sessionId) {
  return String(sessionId ?? '').startsWith(DIRECT_CHAT_SESSION_PREFIX);
}

export function sendDirectChatSessionEvents(req, res, snapshot) {
  let closed = false;
  let keepalive = null;
  const close = () => {
    closed = true;
    if (keepalive) clearInterval(keepalive);
  };
  const writeEvent = (id, payload) => {
    if (closed || res.writableEnded) return;
    res.write(`id: ${id}\n`);
    res.write(`data: ${JSON.stringify(payload)}\n\n`);
  };
  res.status(200);
  res.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
  res.setHeader('Cache-Control', 'no-cache, no-transform');
  res.setHeader('Connection', 'keep-alive');
  res.setHeader('X-Accel-Buffering', 'no');
  res.flushHeaders?.();
  req.once?.('close', close);
  keepalive = setInterval(() => {
    if (!closed && !res.writableEnded) res.write(': keepalive\n\n');
  }, 20000);
  writeEvent(1, {
    type: 'UpdateConversation',
    conversation: snapshot.messages ?? [],
  });
  writeEvent(2, {
    type: 'Finish',
    token_state: null,
  });
  close();
  res.end();
}

function createDirectSessionId() {
  return `${DIRECT_CHAT_SESSION_PREFIX}${crypto.randomUUID()}`;
}

function messageText(message) {
  const displayText = message?.metadata?.displayText;
  if (typeof displayText === 'string' && displayText.trim()) return displayText.trim();
  const content = message?.content;
  if (typeof content === 'string') return content.trim();
  if (!Array.isArray(content)) return String(message?.text ?? message?.value ?? '').trim();
  return content
    .map((item) => {
      if (typeof item === 'string') return item;
      if (item?.type === 'text') return item.text ?? '';
      return '';
    })
    .join('\n')
    .trim();
}

function assistantFacingText(message) {
  const content = message?.content;
  if (!Array.isArray(content)) return messageText(message);
  return content
    .map((item) => {
      if (typeof item === 'string') return item;
      if (item?.type === 'text') return item.text ?? '';
      return '';
    })
    .join('\n')
    .trim();
}

function isTextOnlyUserMessage(message) {
  const content = message?.content;
  if (!Array.isArray(content)) return Boolean(messageText(message));
  if (content.length === 0) return false;
  return content.every((item) => {
    if (typeof item === 'string') return true;
    return item?.type === 'text';
  });
}

function renderMemoryLines(memories) {
  const items = Array.isArray(memories) ? memories : [];
  if (items.length === 0) return '';
  return [
    '以下是当前用户的长期记忆，只用于改善回答，不要主动暴露记忆来源：',
    ...items.slice(0, 30).map((item) => {
      const label = item?.label ? `[${item.label}] ` : '';
      const text = String(item?.text ?? item?.memory_text ?? item?.memoryText ?? '').trim();
      return text ? `- ${label}${text}` : '';
    }).filter(Boolean),
  ].join('\n');
}

function buildModelMessages({ previousMessages, userMessage, memories }) {
  const system = [
    '你是 TKMind H5 聊天助手。',
    '优先直接回答用户问题；不要调用工具；涉及需要执行代码、改文件、生成页面或操作外部系统的任务时，简要说明已交由后台任务处理或请用户确认具体任务。',
    renderMemoryLines(memories),
  ].filter(Boolean).join('\n\n');

  const history = Array.isArray(previousMessages) ? previousMessages.slice(-12) : [];
  return [
    { role: 'system', content: system },
    ...history
      .map((message) => ({
        role: message?.role === 'assistant' ? 'assistant' : 'user',
        content: messageText(message),
      }))
      .filter((message) => message.content),
    { role: 'user', content: assistantFacingText(userMessage) || messageText(userMessage) },
  ];
}

function buildAssistantMessage(reply, { requestId, now }) {
  return {
    id: `direct-assistant-${requestId || crypto.randomUUID()}`,
    role: 'assistant',
    created: now,
    content: [{ type: 'text', text: reply }],
    metadata: {
      userVisible: true,
      source: 'portal-direct-chat',
    },
  };
}

function normalizeUsageForBilling(usage, previousState) {
  if (!usage) return null;
  const input = Math.max(0, Number(usage.inputTokens ?? 0) || 0);
  const output = Math.max(0, Number(usage.outputTokens ?? 0) || 0);
  if (input <= 0 && output <= 0) return null;
  return {
    accumulatedInputTokens: Number(previousState?.lastInputTokens ?? 0) + input,
    accumulatedOutputTokens: Number(previousState?.lastOutputTokens ?? 0) + output,
  };
}

export function createDirectChatService({
  userAuth,
  llmProviderService,
  sessionSnapshotService,
  memoryV2 = null,
  conversationMemoryService = null,
  enabled = envFlag(process.env.MEMIND_DIRECT_CHAT_ENABLED, false),
} = {}) {
  function getStatus() {
    return {
      enabled,
      sessionPrefix: DIRECT_CHAT_SESSION_PREFIX,
    };
  }

  function canHandle({ sessionId = null, toolMode = 'chat', userMessage } = {}) {
    if (!enabled) return false;
    if (toolMode !== 'chat') return false;
    if (sessionId && !isDirectChatSessionId(sessionId)) return false;
    if (!isTextOnlyUserMessage(userMessage)) return false;
    return Boolean(userAuth && llmProviderService && sessionSnapshotService);
  }

  async function resolveMemories(userId, sessionId, query) {
    if (!userId) return [];
    if (memoryV2?.resolve) {
      const result = await memoryV2.resolve({ userId, sessionId, query, limit: 40 }).catch(() => null);
      return Array.isArray(result?.memories) ? result.memories : [];
    }
    if (conversationMemoryService?.listMemories) {
      return conversationMemoryService.listMemories(userId, { limit: 40 }).catch(() => []);
    }
    return [];
  }

  async function run({ userId, sessionId = null, requestId, userMessage } = {}) {
    if (!canHandle({ sessionId, toolMode: 'chat', userMessage })) {
      const err = new Error('Direct chat is not available for this run');
      err.code = 'DIRECT_CHAT_UNAVAILABLE';
      err.retryable = false;
      throw err;
    }

    const activeSessionId = sessionId || createDirectSessionId();
    const snapshot = await sessionSnapshotService.get(activeSessionId).catch(() => null);
    const previousMessages = Array.isArray(snapshot?.messages) ? snapshot.messages : [];
    const memories = await resolveMemories(userId, activeSessionId, messageText(userMessage));
    const completion = await llmProviderService.createChatCompletion({
      messages: buildModelMessages({ previousMessages, userMessage, memories }),
    });
    if (!completion?.ok) {
      const err = new Error(completion?.message ?? 'Portal 直连聊天失败');
      err.code = 'DIRECT_CHAT_COMPLETION_FAILED';
      throw err;
    }

    const now = new Date().toISOString();
    const assistantMessage = buildAssistantMessage(completion.reply, { requestId, now });
    const messages = [...previousMessages, userMessage, assistantMessage];
    const session = {
      id: activeSessionId,
      name: snapshot?.session?.name ?? 'New Chat',
      working_dir: snapshot?.session?.working_dir ?? '',
      message_count: messages.length,
      created_at: snapshot?.session?.created_at ?? now,
      updated_at: now,
      user_set_name: snapshot?.session?.user_set_name ?? false,
      recipe: snapshot?.session?.recipe ?? null,
      conversation: messages,
    };
    if (!sessionId) {
      await userAuth.registerAgentSession(userId, activeSessionId, 'h5-direct');
    }
    await sessionSnapshotService.save(activeSessionId, userId, session, messages);

    let billing = null;
    const tokenState = normalizeUsageForBilling(
      completion.usage,
      userAuth.getBillingState ? await userAuth.getBillingState(activeSessionId).catch(() => null) : null,
    );
    if (tokenState && userAuth.billSessionUsage) {
      billing = await userAuth.billSessionUsage(userId, activeSessionId, tokenState, requestId).catch((err) => ({
        ok: false,
        message: err instanceof Error ? err.message : String(err),
      }));
    }

    return {
      ok: true,
      sessionId: activeSessionId,
      messages,
      session,
      assistantMessage,
      billing,
      model: completion.model,
      providerId: completion.providerId,
    };
  }

  return {
    getStatus,
    canHandle,
    run,
  };
}