memind/direct-chat-service.mjs

import crypto from 'node:crypto';
import { MEMORY_INTERVENTION_LIMIT } from './memory-intervention.mjs';
import { resolveMemoriesWithLegacyFallback } from './memory-legacy-fallback.mjs';
import { isMemoryRecallQuestion } from './chat-intent-router.mjs';
import { resolveSessionAccess } from './session-broker.mjs';

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);
}

const TASK_EXECUTION_REQUIRED_PATTERNS = [
  /\b(public\/[^\s"'<>]+\.html)\b/i,
  /\b(html|h5|web\s?page|landing\s?page|microsite|docx|word)\b/i,
  /\b(write|create|generate|publish|download|export|save|edit)\b.{0,40}\b(file|page|html|docx|word|asset)\b/i,
  /\b(file|page|html|docx|word|asset)\b.{0,40}\b(write|create|generate|publish|download|export|save|edit)\b/i,
  /(?:生成|创建|制作|做|写|设计|发布|导出|下载|保存|修改|编辑).{0,24}(?:页面|网页|HTML|html|H5|h5|文件|文档|Word|word|docx|下载页|公开页|分享页|落地页|活动页)/u,
  /(?:页面|网页|HTML|html|H5|h5|文件|文档|Word|word|docx|下载页|公开页|分享页|落地页|活动页).{0,24}(?:生成|创建|制作|做|写|设计|发布|导出|下载|保存|修改|编辑)/u,
  /(?:写|做|生成|制作|设计).{0,16}主题(?:页面|文章)/u,
  /主题(?:页面|文章).{0,16}(?:写|做|生成|制作|设计)/u,
  /MindSpace\/[^/\s]+\/public\/[^\s"'<>]+\.html/i,
];

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 requiresTaskExecution(message) {
  const text = `${messageText(message)}\n${assistantFacingText(message)}`.trim();
  return TASK_EXECUTION_REQUIRED_PATTERNS.some((pattern) => pattern.test(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 buildMemorySystemBlock({ memories, routingMemoryContent } = {}) {
  const routed = String(routingMemoryContent ?? '').trim();
  if (routed) {
    return [
      '以下是当前用户的长期记忆，只用于改善回答，不要主动暴露记忆来源：',
      routed,
    ].join('\n');
  }
  return renderMemoryLines(memories);
}

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

function evaluateCanHandle({
  sessionId = null,
  toolMode = 'chat',
  userMessage,
  routingDecision = null,
  enabled: enabledOverride,
  userAuth,
  llmProviderService,
  sessionSnapshotService,
} = {}) {
  const resolvedEnabled = enabledOverride ?? true;
  if (!resolvedEnabled) return { ok: false, reason: 'disabled' };
  if (toolMode !== 'chat') return { ok: false, reason: 'tool_mode_not_chat' };
  if (routingDecision === 'agent_orchestration') {
    return { ok: false, reason: 'routing_agent_orchestration' };
  }
  if (
    sessionId &&
    !isDirectChatSessionId(sessionId) &&
    routingDecision !== 'direct_chat'
  ) {
    return { ok: false, reason: 'session_not_direct_chat' };
  }
  if (!isTextOnlyUserMessage(userMessage)) {
    return { ok: false, reason: 'non_text_message' };
  }
  if (routingDecision !== 'direct_chat' && requiresTaskExecution(userMessage)) {
    return { ok: false, reason: 'task_execution_required' };
  }
  if (!userAuth || !llmProviderService || !sessionSnapshotService) {
    return { ok: false, reason: 'dependencies_unavailable' };
  }
  return { ok: true, reason: null };
}

export function isPortalDirectChatSnapshot(snapshot, { sessionId = null } = {}) {
  if (!snapshot) return false;
  const messages = Array.isArray(snapshot?.messages) ? snapshot.messages : [];
  const lastAssistant = [...messages].reverse().find((message) => message?.role === 'assistant');
  if (lastAssistant?.metadata?.source !== 'portal-direct-chat') return false;
  // Mixed agent sessions: a user turn after the portal-direct-chat assistant means
  // execution moved to goosed — never replay the stale direct-chat snapshot.
  if (sessionId && !isDirectChatSessionId(sessionId)) {
    const lastAssistantIndex = messages.indexOf(lastAssistant);
    const hasUserTurnAfterDirectChat = messages
      .slice(lastAssistantIndex + 1)
      .some((message) => message?.role === 'user');
    if (hasUserTurnAfterDirectChat) return false;
  }
  const updatedAtRaw = snapshot?.session?.updated_at ?? snapshot?.meta?.saved_at ?? 0;
  const updatedAt = Date.parse(String(updatedAtRaw)) || Number(updatedAtRaw) || 0;
  if (!updatedAt) return true;
  return Date.now() - updatedAt <= 5 * 60 * 1000;
}

export async function shouldExpirePortalDirectChatSnapshot(pool, sessionId, snapshot) {
  if (!pool || !sessionId || !snapshot) return false;
  if (isDirectChatSessionId(sessionId)) return false;
  if (!isPortalDirectChatSnapshot(snapshot, { sessionId })) return false;
  const syncedAt = Number(snapshot?.meta?.synced_at ?? 0);
  if (!syncedAt) return false;
  const [rows] = await pool.query(
    `SELECT r.id
       FROM h5_agent_runs r
      WHERE r.agent_session_id = ?
        AND r.status = 'succeeded'
        AND r.completed_at > ?
        AND NOT EXISTS (
          SELECT 1
            FROM h5_agent_run_events e
           WHERE e.run_id = r.id
             AND e.event_type = 'direct_chat_completed'
        )
      LIMIT 1`,
    [sessionId, syncedAt],
  );
  return rows.length > 0;
}

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,
  sessionAccess = null,
  llmProviderService,
  sessionSnapshotService,
  memoryV2 = null,
  conversationMemoryService = null,
  enabled = envFlag(process.env.MEMIND_DIRECT_CHAT_ENABLED, true),
} = {}) {
  const sessionStore = resolveSessionAccess({ userAuth, sessionAccess });
  function getStatus() {
    return {
      enabled,
      sessionPrefix: DIRECT_CHAT_SESSION_PREFIX,
    };
  }

  function canHandle(input = {}) {
    return evaluateCanHandle({
      ...input,
      enabled,
      userAuth,
      llmProviderService,
      sessionSnapshotService,
    }).ok;
  }

  function explainCanHandle(input = {}) {
    return evaluateCanHandle({
      ...input,
      enabled,
      userAuth,
      llmProviderService,
      sessionSnapshotService,
    });
  }

  async function resolveMemories(userId, sessionId, query, { limit = MEMORY_INTERVENTION_LIMIT.LIGHT_DIRECT_CHAT } = {}) {
    return resolveMemoriesWithLegacyFallback({
      memoryV2,
      conversationMemoryService,
      userId,
      sessionId,
      query,
      limit,
      recallQuestion: isMemoryRecallQuestion(query),
    });
  }

  async function run({
    userId,
    sessionId = null,
    requestId,
    userMessage,
    routingDecision = null,
    routingMemory = null,
    onSessionReady = null,
  } = {}) {
    if (!canHandle({ sessionId, toolMode: 'chat', userMessage, routingDecision })) {
      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 now = new Date().toISOString();
    if (!sessionId) {
      await sessionStore.registerAgentSession(userId, activeSessionId, 'h5-direct');
    }
    if (typeof onSessionReady === 'function') {
      await onSessionReady(activeSessionId);
    }
    const pendingMessages = [...previousMessages, userMessage];
    await sessionSnapshotService.save(
      activeSessionId,
      userId,
      {
        id: activeSessionId,
        name: snapshot?.session?.name ?? 'New Chat',
        working_dir: snapshot?.session?.working_dir ?? '',
        message_count: pendingMessages.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: pendingMessages,
      },
      pendingMessages,
    );
    const recallQuestion = isMemoryRecallQuestion(messageText(userMessage));
    const routedMemoryContent =
      !recallQuestion &&
      routingMemory && !routingMemory.skipped && !routingMemory.degraded
        ? String(routingMemory.content ?? '').trim()
        : '';
    const memories = routedMemoryContent
      ? []
      : await resolveMemories(userId, activeSessionId, messageText(userMessage));
    const completion = await llmProviderService.createChatCompletion({
      messages: buildModelMessages({
        previousMessages,
        userMessage,
        memories,
        routingMemoryContent: routedMemoryContent,
      }),
    });
    if (!completion?.ok) {
      const err = new Error(completion?.message ?? 'Portal 直连聊天失败');
      err.code = 'DIRECT_CHAT_COMPLETION_FAILED';
      throw err;
    }

    const assistantMessage = buildAssistantMessage(completion.reply, { requestId, now });
    const messages = [...pendingMessages, 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,
    };
    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,
    explainCanHandle,
    run,
  };
}