memind/scheduled-task-executor.mjs

import crypto from 'node:crypto';
import fs from 'node:fs';
import path from 'node:path';
import { buildChatSkillPrompt, SCHEDULED_TASK_AUTOMATION_SKILL_NAME } from './chat-skills.mjs';
import {
  markPageDeliveryContractFailed,
  markPageDeliveryContractReady,
  normalizeDeliveryRelativePath,
  preparePageDeliveryContract,
  releaseMaterializedPageDeliveryContracts,
} from './mindspace-delivery-contract.mjs';
import {
  collectOwnPublicHtmlRelativePaths,
  materializeMissingPublicHtmlWrites,
} from './mindspace-public-finish-sync.mjs';
import { localDateLabel } from './schedule-time.mjs';
import { executeCursorChannelCodeRun } from './wechat-cursor-agent-run.mjs';
import { resolveCursorScheduledTaskEligible } from './wechat-cursor-executor-policy.mjs';

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

export function buildScheduledTaskExecutionPrompt(task, {
  now = Date.now(),
  timezone = task?.timezone ?? 'Asia/Shanghai',
} = {}) {
  const dateLabel = localDateLabel(now, timezone);
  const body = [
    '【定时任务自动执行 Scheduled Automation】',
    `任务标题：${task.title}`,
    `执行日期：${dateLabel}`,
    `任务要求：${task.taskSpec}`,
    '执行约束：',
    '- 这是系统自动触发的定时任务，请直接完成可交付结果，不要反问用户。',
    '- 禁止向用户追问时间、频率或任务内容；taskSpec 已是完整执行说明。',
    '- 禁止调用 scheduled_task_create / scheduled_task_list / scheduled_task_cancel；只执行 taskSpec。',
    '- 若需要生成页面，必须先 load_skill → static-page-publish，再 write_file 到 public/*.html，并给出正式可访问 URL。',
    '- 若只需摘要/文本，给出完整中文结果摘要。',
    '- 完成后在回复中明确写出交付结果（链接或摘要）。',
  ].join('\n');
  const prefix = buildChatSkillPrompt(
    'scheduled-task-automation',
    SCHEDULED_TASK_AUTOMATION_SKILL_NAME,
  );
  return {
    role: 'user',
    content: [{ type: 'text', text: `${prefix}${body}` }],
    metadata: {
      displayText: `定时任务：${task.title}`,
      userVisible: false,
      agentVisible: true,
      memindScheduledTask: {
        taskId: task.id,
        recurrence: task.recurrence,
        automated: true,
      },
    },
  };
}

export function extractScheduledTaskDeliveryText(messages, task) {
  const lastAssistant = [...(Array.isArray(messages) ? messages : [])]
    .reverse()
    .find((message) => message?.role === 'assistant');
  const text = messageText(lastAssistant);
  if (text) return text.trim();
  return `定时任务「${task.title}」已执行完成。`;
}

export function formatScheduledTaskDeliveryMessage(task, deliveryText) {
  const header = `定时任务完成：${task.title}`;
  const body = String(deliveryText ?? '').trim() || '任务已执行，请前往 MindSpace 查看结果。';
  return `${header}\n\n${body}`.trim();
}

export function resolveScheduledTaskPublicBaseUrl(env = process.env) {
  return String(env.H5_PUBLIC_BASE_URL ?? 'https://m.tkmind.cn').replace(/\/+$/, '');
}

export function buildScheduledTaskVerifiedHtmlUrls(userId, readyPaths = [], {
  publicBaseUrl = resolveScheduledTaskPublicBaseUrl(),
} = {}) {
  const normalizedUserId = String(userId ?? '').trim();
  if (!normalizedUserId) return [];
  return [...new Set(
    (Array.isArray(readyPaths) ? readyPaths : [])
      .map((relativePath) => normalizeDeliveryRelativePath(relativePath))
      .filter(Boolean)
      .map((relativePath) => `${publicBaseUrl}/MindSpace/${normalizedUserId}/${relativePath}`),
  )];
}

const SCHEDULED_TASK_CLARIFICATION_PATTERNS = [
  /需确认/u,
  /请确认/u,
  /请问/u,
  /未指定/u,
  /需要澄清/u,
  /在创建前需要确认/u,
  /信息不完整/u,
  /具体几点/u,
  /缺(?:少|失)/u,
];

const DEFAULT_SCHEDULED_TASK_DELIVERY_POLL_INTERVAL_MS = 5_000;

export function resolveScheduledTaskDeliveryPollIntervalMs(
  env = process.env,
) {
  const parsed = Number(
    env.H5_SCHEDULED_TASK_DELIVERY_POLL_INTERVAL_MS
      ?? DEFAULT_SCHEDULED_TASK_DELIVERY_POLL_INTERVAL_MS,
  );
  return Number.isFinite(parsed) && parsed > 0
    ? parsed
    : DEFAULT_SCHEDULED_TASK_DELIVERY_POLL_INTERVAL_MS;
}

/** @deprecated use resolveScheduledTaskDeliveryPollIntervalMs */
export function resolveScheduledTaskDeliveryRetryDelaysMs(env = process.env) {
  const interval = resolveScheduledTaskDeliveryPollIntervalMs(env);
  return [0, interval, interval, interval];
}

export function extractPublicHtmlPathsFromText(text) {
  const paths = new Set();
  const normalized = String(text ?? '');
  for (const match of normalized.matchAll(
    /public\/[^\s"'<>]+\.html/gi,
  )) {
    paths.add(match[0].replace(/\\/g, '/'));
  }
  for (const match of normalized.matchAll(
    /\/MindSpace\/[^/\s"'<>]+\/(public\/[^\s"'<>]+\.html)/gi,
  )) {
    paths.add(match[1].replace(/\\/g, '/'));
  }
  return [...paths];
}

export function deliveryTextPromisesPublicHtml(text) {
  const normalized = String(text ?? '').trim();
  if (!normalized) return false;
  if (extractPublicHtmlPathsFromText(normalized).length > 0) return true;
  return /\/MindSpace\/[^/\s"'<>]+/i.test(normalized)
    && /\.html/i.test(normalized);
}

export function looksLikeScheduledTaskNonDelivery(text, { readyPaths = [] } = {}) {
  if (Array.isArray(readyPaths) && readyPaths.length > 0) return false;
  const normalized = String(text ?? '').trim();
  if (!normalized) return true;
  if (SCHEDULED_TASK_CLARIFICATION_PATTERNS.some((pattern) => pattern.test(normalized))) {
    return true;
  }
  if (deliveryTextPromisesPublicHtml(normalized)) return true;
  if (/https?:\/\//i.test(normalized)) return false;
  if (/public\/[^\s]+\.html/i.test(normalized)) return false;
  if (/页面链接/u.test(normalized)) return false;
  if (/(?:已生成|已完成|交付).{0,24}(?:页面|链接|结果)/u.test(normalized)) return false;
  if (/^验证成功[。!！]?$/u.test(normalized)) return false;
  if (/^(?:任务)?(?:已)?完成[。!！]?$/u.test(normalized)) return false;
  if (/^【验证】/u.test(normalized) && normalized.length >= 6) return false;
  return normalized.length < 12;
}

async function refreshScheduledTaskMessages({
  userId,
  sessionId,
  tkmindProxy,
  sessionSnapshotService,
}) {
  if (typeof tkmindProxy?.fetchSessionConversationForUser === 'function') {
    return tkmindProxy
      .fetchSessionConversationForUser(userId, sessionId)
      .catch(() => []);
  }
  if (typeof sessionSnapshotService?.get === 'function') {
    const snapshot = await sessionSnapshotService.get(sessionId).catch(() => null);
    return snapshot?.messages ?? snapshot?.conversation?.messages ?? [];
  }
  return [];
}

function collectScheduledTaskPageRelativePaths({
  messages,
  publishDir,
  currentUser,
  userId,
  deliveryText = '',
}) {
  const materialized = materializeMissingPublicHtmlWrites({
    messages,
    publishDir,
  });
  const relativePaths = new Set(
    collectOwnPublicHtmlRelativePaths({
      messages,
      currentUser: currentUser ?? { id: userId },
      publishDir,
      materialized: materialized.materialized,
      skipped: materialized.skipped,
    }),
  );
  for (const relativePath of extractPublicHtmlPathsFromText(deliveryText)) {
    relativePaths.add(relativePath);
  }
  return {
    materialized,
    relativePaths: [...relativePaths],
  };
}

export async function finalizeScheduledTaskPageDelivery({
  pool,
  userId,
  sessionId,
  messages,
  publishDir,
  currentUser = null,
  deliveryText = '',
  logger = console,
} = {}) {
  if (!pool || !userId || !publishDir) return [];

  const { relativePaths } = collectScheduledTaskPageRelativePaths({
    messages,
    publishDir,
    currentUser,
    userId,
    deliveryText,
  });
  const pathsToRelease = new Set(relativePaths);

  if (sessionId) {
    const [rows] = await pool.query(
      `SELECT workspace_relative_path
       FROM h5_page_delivery_contracts
       WHERE user_id = ? AND status = 'preparing'`,
      [userId],
    );
    for (const row of rows ?? []) {
      if (row?.workspace_relative_path) {
        pathsToRelease.add(row.workspace_relative_path);
      }
    }
  }

  for (const relativePath of pathsToRelease) {
    await preparePageDeliveryContract({
      pool,
      userId,
      requestId: sessionId,
      relativePath,
      pgRequired: false,
    }).catch((error) => {
      logger.warn?.(
        `[ScheduledTask] prepare delivery contract failed for ${relativePath}:`,
        error,
      );
    });
  }

  const readyPaths = await releaseMaterializedPageDeliveryContracts({
    pool,
    userId,
    relativePaths: [...pathsToRelease],
    allowPgRequired: true,
    publishDir,
  }).catch((error) => {
    logger.warn?.('[ScheduledTask] release delivery contracts failed:', error);
    return [];
  });
  for (const relativePath of readyPaths) {
    logger.info?.('[ScheduledTask] delivery contract ready', {
      userId,
      sessionId,
      relativePath,
    });
  }
  return readyPaths;
}

export async function awaitScheduledTaskPageDelivery({
  pool,
  userId,
  sessionId,
  messages,
  publishDir,
  deliveryText = '',
  task = null,
  tkmindProxy = null,
  sessionSnapshotService = null,
  timeoutMs = 15 * 60 * 1000,
  pollIntervalMs = resolveScheduledTaskDeliveryPollIntervalMs(),
  sleepFn = (delayMs) => new Promise((resolve) => {
    setTimeout(resolve, delayMs);
  }),
  logger = console,
} = {}) {
  let currentMessages = Array.isArray(messages) ? messages : [];
  let currentDeliveryText = String(deliveryText ?? '');
  let readyPaths = [];
  const deadline = Date.now() + Math.max(Number(pollIntervalMs) || 0, Number(timeoutMs) || 0);
  let attempt = 0;

  while (true) {
    attempt += 1;
    readyPaths = await finalizeScheduledTaskPageDelivery({
      pool,
      userId,
      sessionId,
      messages: currentMessages,
      publishDir,
      deliveryText: currentDeliveryText,
      logger,
    }).catch((error) => {
      logger.warn?.('[ScheduledTask] finalize page delivery failed:', error);
      return [];
    });

    const promisesHtml = deliveryTextPromisesPublicHtml(currentDeliveryText)
      || collectScheduledTaskPageRelativePaths({
        messages: currentMessages,
        publishDir,
        userId,
        deliveryText: currentDeliveryText,
      }).relativePaths.length > 0;
    if (!promisesHtml || readyPaths.length > 0 || Date.now() >= deadline) {
      if (promisesHtml && readyPaths.length === 0 && Date.now() >= deadline) {
        logger.warn?.('[ScheduledTask] page delivery timed out while preparing', {
          userId,
          sessionId,
          attempt,
          timeoutMs,
        });
      }
      break;
    }

    logger.warn?.('[ScheduledTask] page delivery still preparing', {
      userId,
      sessionId,
      attempt,
      nextPollMs: pollIntervalMs,
    });
    await sleepFn(pollIntervalMs);
    currentMessages = await refreshScheduledTaskMessages({
      userId,
      sessionId,
      tkmindProxy,
      sessionSnapshotService,
    });
    if (task) {
      currentDeliveryText = extractScheduledTaskDeliveryText(currentMessages, task);
    }
  }

  return {
    messages: currentMessages,
    readyPaths,
  };
}

export async function reconcileStuckStaticPageDeliveryContracts({
  pool,
  h5Root,
  limit = 20,
  logger = console,
} = {}) {
  if (!pool || !h5Root) return [];
  const [rows] = await pool.query(
    `SELECT user_id, workspace_relative_path
     FROM h5_page_delivery_contracts
     WHERE status = 'preparing' AND data_mode = 'static'
     ORDER BY updated_at ASC
     LIMIT ?`,
    [Math.max(1, Number(limit) || 20)],
  );
  const released = [];
  for (const row of rows ?? []) {
    const relativePath = normalizeDeliveryRelativePath(
      row.workspace_relative_path,
    );
    if (!relativePath) continue;
    const filePath = path.join(
      h5Root,
      'MindSpace',
      row.user_id,
      relativePath,
    );
    if (!fs.existsSync(filePath)) {
      if (
        await markPageDeliveryContractFailed({
          pool,
          userId: row.user_id,
          relativePath,
          failureReason: 'materialized_html_missing',
        })
      ) {
        logger.info?.('[ScheduledTask] failed orphan static delivery contract', {
          userId: row.user_id,
          relativePath,
        });
      }
      continue;
    }
    if (
      await markPageDeliveryContractReady({
        pool,
        userId: row.user_id,
        relativePath,
      })
    ) {
      released.push({ userId: row.user_id, relativePath });
      logger.info?.('[ScheduledTask] reconciled static delivery contract', {
        userId: row.user_id,
        relativePath,
      });
    }
  }
  return released;
}

export async function resendScheduledTaskWechatForReadyPage({
  pool,
  userId,
  relativePath,
  notificationDispatcher = null,
  publicBaseUrl = resolveScheduledTaskPublicBaseUrl(),
  lookbackMs = 24 * 60 * 60 * 1000,
  now = Date.now(),
  logger = console,
} = {}) {
  const normalizedPath = normalizeDeliveryRelativePath(relativePath);
  const normalizedUserId = String(userId ?? '').trim();
  if (
    !pool
    || !normalizedUserId
    || !normalizedPath
    || typeof notificationDispatcher?.sendScheduleNotification !== 'function'
  ) {
    return false;
  }

  const [rows] = await pool.query(
    `SELECT id, title, notify_channel, last_result_json, last_run_at
     FROM h5_scheduled_tasks
     WHERE user_id = ? AND last_run_at IS NOT NULL AND last_run_at >= ?
     ORDER BY last_run_at DESC
     LIMIT 20`,
    [normalizedUserId, now - Math.max(lookbackMs, 60_000)],
  );

  const verifiedUrl = `${publicBaseUrl}/MindSpace/${normalizedUserId}/${normalizedPath}`;
  for (const row of rows ?? []) {
    const channel = row.notify_channel ?? 'both';
    if (channel !== 'wechat' && channel !== 'both') continue;

    const lastResult = row.last_result_json && typeof row.last_result_json === 'object'
      ? row.last_result_json
      : null;
    const deliveryText = String(lastResult?.deliveryText ?? '');
    if (
      !deliveryText.includes(normalizedPath)
      && !deliveryText.includes(verifiedUrl)
    ) {
      continue;
    }

    const priorDelivery = lastResult?.wechatDelivery;
    if (
      priorDelivery?.sentAt
      && Array.isArray(priorDelivery.relativePaths)
      && priorDelivery.relativePaths.includes(normalizedPath)
    ) {
      continue;
    }

    const text = formatScheduledTaskDeliveryMessage(
      { title: row.title },
      deliveryText,
    );
    const verifiedHtmlUrls = buildScheduledTaskVerifiedHtmlUrls(
      normalizedUserId,
      [normalizedPath],
      { publicBaseUrl },
    );
    const sent = await notificationDispatcher.sendScheduleNotification({
      userId: normalizedUserId,
      text,
      verifiedHtmlUrls,
    }).catch((error) => {
      logger.warn?.('[ScheduledTask] reconcile wechat resend failed:', error);
      return false;
    });
    if (!sent) continue;

    const nextResult = {
      ...(lastResult ?? {}),
      wechatDelivery: {
        sentAt: now,
        relativePaths: [normalizedPath],
        source: 'reconcile',
      },
    };
    await pool.query(
      `UPDATE h5_scheduled_tasks
       SET last_result_json = ?, updated_at = ?
       WHERE id = ? AND user_id = ?`,
      [JSON.stringify(nextResult), now, row.id, normalizedUserId],
    );
    logger.info?.('[ScheduledTask] reconciled wechat delivery resent', {
      userId: normalizedUserId,
      taskId: row.id,
      relativePath: normalizedPath,
    });
    return true;
  }

  return false;
}

export async function executeScheduledTask(task, {
  userAuth,
  tkmindProxy,
  agentRunGateway = null,
  cursorExecutorPolicyService = null,
  sessionSnapshotService = null,
  pool = null,
  h5Root = null,
  timeoutMs = 15 * 60 * 1000,
  logger = console,
} = {}) {
  if (!task?.userId) throw new Error('缺少 task.userId');
  if (!userAuth || typeof userAuth.canUseChat !== 'function') {
    throw new Error('缺少 userAuth.canUseChat');
  }

  const gate = await userAuth.canUseChat(task.userId);
  if (!gate?.ok) {
    const err = new Error(gate?.message ?? '当前账户无法执行定时任务');
    err.code = gate?.code ?? 'CHAT_NOT_ALLOWED';
    throw err;
  }

  const requestId = crypto.randomUUID();
  const userMessage = buildScheduledTaskExecutionPrompt(task);
  const publishDir = h5Root && task.userId
    ? path.join(h5Root, 'MindSpace', task.userId)
    : null;

  let cursorPolicy = null;
  let useCursorPath = false;
  if (agentRunGateway && cursorExecutorPolicyService?.getEffectivePolicy) {
    try {
      cursorPolicy = await cursorExecutorPolicyService.getEffectivePolicy(task.userId, {
        userId: task.userId,
      });
      useCursorPath = resolveCursorScheduledTaskEligible({
        user: { userId: task.userId },
        policy: cursorPolicy,
      });
    } catch (err) {
      logger.warn?.('[ScheduledTask] cursor policy lookup failed:', err);
    }
  }

  let sessionId = null;
  let messages = [];

  if (useCursorPath) {
    logger.info?.('[ScheduledTask] executing via cursor channel', {
      taskId: task.id,
      userId: task.userId,
      requestId,
    });
    try {
      const cursorResult = await executeCursorChannelCodeRun({
        agentRunGateway,
        userId: task.userId,
        requestId,
        displayText: `定时任务：${task.title}`,
        agentPrompt: userMessage.content[0]?.text ?? userMessage.content,
        intentKind: 'page.generate',
        channel: 'scheduled_task',
        taskType: 'h5_chat_code_task',
        policy: cursorPolicy,
        forceCursorExecutor: true,
        timeoutMs,
        logger,
      });
      messages = cursorResult.messages ?? [];
      sessionId = cursorResult.runId ?? null;
    } catch (cursorErr) {
      if (cursorPolicy?.fallbackToDeepseek === false) throw cursorErr;
      logger.warn?.('[ScheduledTask] cursor execution failed, falling back to goose:', cursorErr);
      useCursorPath = false;
    }
  }

  if (!useCursorPath) {
    if (
      !tkmindProxy
      || typeof tkmindProxy.startSessionForUser !== 'function'
      || typeof tkmindProxy.submitSessionReplyAndAwaitFinishForUser !== 'function'
    ) {
      throw new Error('缺少 tkmindProxy 会话执行能力');
    }
    const started = await tkmindProxy.startSessionForUser(task.userId, {
      origin: 'h5',
    });
    sessionId = started?.id ?? started?.sessionId;
    if (!sessionId) throw new Error('创建定时任务会话失败');

    logger.info?.('[ScheduledTask] executing via goose session', {
      taskId: task.id,
      userId: task.userId,
      sessionId,
      requestId,
    });

    await tkmindProxy.submitSessionReplyAndAwaitFinishForUser(
      task.userId,
      sessionId,
      requestId,
      userMessage,
      { timeoutMs },
    );

    messages = await refreshScheduledTaskMessages({
      userId: task.userId,
      sessionId,
      tkmindProxy,
      sessionSnapshotService,
    });
  }

  let deliveryText = extractScheduledTaskDeliveryText(messages, task);
  const deliveryResult = publishDir
    ? await awaitScheduledTaskPageDelivery({
      pool,
      userId: task.userId,
      sessionId,
      messages,
      publishDir,
      deliveryText,
      task,
      tkmindProxy,
      sessionSnapshotService,
      timeoutMs,
      logger,
    })
    : { messages, readyPaths: [] };
  messages = deliveryResult.messages;
  const readyPaths = deliveryResult.readyPaths;

  deliveryText = extractScheduledTaskDeliveryText(messages, task);
  if (
    readyPaths.length > 0
    && !/https?:\/\//i.test(deliveryText)
    && task.userId
  ) {
    const links = readyPaths.map(
      (relativePath) => `${resolveScheduledTaskPublicBaseUrl()}/MindSpace/${task.userId}/${relativePath}`,
    );
    deliveryText = `${deliveryText}\n\n页面链接：\n${links.join('\n')}`.trim();
  }

  return {
    sessionId,
    requestId,
    deliveryText,
    messages,
    readyPaths,
    executor: useCursorPath ? 'cursor' : 'goose',
  };
}