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'; 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, 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'); } if ( !tkmindProxy || typeof tkmindProxy.startSessionForUser !== 'function' || typeof tkmindProxy.submitSessionReplyAndAwaitFinishForUser !== 'function' ) { throw new Error('缺少 tkmindProxy 会话执行能力'); } 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 started = await tkmindProxy.startSessionForUser(task.userId, { origin: 'h5', }); const sessionId = started?.id ?? started?.sessionId; if (!sessionId) throw new Error('创建定时任务会话失败'); const userMessage = buildScheduledTaskExecutionPrompt(task); logger.info?.('[ScheduledTask] executing', { taskId: task.id, userId: task.userId, sessionId, requestId, }); await tkmindProxy.submitSessionReplyAndAwaitFinishForUser( task.userId, sessionId, requestId, userMessage, { timeoutMs }, ); let messages = await refreshScheduledTaskMessages({ userId: task.userId, sessionId, tkmindProxy, sessionSnapshotService, }); const publishDir = h5Root && task.userId ? path.join(h5Root, 'MindSpace', task.userId) : null; 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, }; }