fix(scheduled-task): release page delivery contracts before notifying users

Scheduled automation now prepares and retries MindSpace delivery contracts, blocks WeChat pushes when public HTML links are not ready, and reconciles stuck static preparing contracts on each worker scan. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-16 08:26:21 +08:00
parent 2b58cdc2c8
commit 7c0ed58ae1
4 changed files with 429 additions and 30 deletions
@@ -1,7 +1,13 @@
 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 { releaseMaterializedPageDeliveryContracts } from './mindspace-delivery-contract.mjs';
+import {
+  markPageDeliveryContractReady,
+  normalizeDeliveryRelativePath,
+  preparePageDeliveryContract,
+  releaseMaterializedPageDeliveryContracts,
+} from './mindspace-delivery-contract.mjs';
 import {
  collectOwnPublicHtmlRelativePaths,
  materializeMissingPublicHtmlWrites,
@@ -83,6 +89,56 @@ const SCHEDULED_TASK_CLARIFICATION_PATTERNS = [
  /缺(?:少|失)/u,
 ];

+const DEFAULT_SCHEDULED_TASK_DELIVERY_RETRY_DELAYS_MS = [
+  250,
+  1_000,
+  3_000,
+  5_000,
+  10_000,
+  30_000,
+  60_000,
+];
+
+export function resolveScheduledTaskDeliveryRetryDelaysMs(
+  env = process.env,
+) {
+  const raw = String(
+    env.H5_SCHEDULED_TASK_DELIVERY_RETRY_DELAYS_MS ?? '',
+  ).trim();
+  if (!raw) return [...DEFAULT_SCHEDULED_TASK_DELIVERY_RETRY_DELAYS_MS];
+  const parsed = raw
+    .split(',')
+    .map((value) => Number(value.trim()))
+    .filter((value) => Number.isFinite(value) && value >= 0);
+  return parsed.length > 0
+    ? parsed
+    : [...DEFAULT_SCHEDULED_TASK_DELIVERY_RETRY_DELAYS_MS];
+}
+
+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();
@@ -90,6 +146,7 @@ export function looksLikeScheduledTaskNonDelivery(text, { readyPaths = [] } = {}
  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;
@@ -100,18 +157,35 @@ export function looksLikeScheduledTaskNonDelivery(text, { readyPaths = [] } = {}
  return normalized.length < 12;
 }

-export async function finalizeScheduledTaskPageDelivery({
-  pool,
+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 = null,
-  logger = console,
-} = {}) {
-  if (!pool || !userId || !publishDir) return [];
-
-  const materialized = materializeMissingPublicHtmlWrites({ messages, publishDir });
+  currentUser,
+  userId,
+  deliveryText = '',
+}) {
+  const materialized = materializeMissingPublicHtmlWrites({
+    messages,
+    publishDir,
+  });
  const relativePaths = new Set(
    collectOwnPublicHtmlRelativePaths({
      messages,
@@ -121,6 +195,35 @@ export async function finalizeScheduledTaskPageDelivery({
      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(
@@ -130,14 +233,31 @@ export async function finalizeScheduledTaskPageDelivery({
      [userId],
    );
    for (const row of rows ?? []) {
-      if (row?.workspace_relative_path) relativePaths.add(row.workspace_relative_path);
+      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: [...relativePaths],
+    relativePaths: [...pathsToRelease],
    allowPgRequired: true,
  }).catch((error) => {
    logger.warn?.('[ScheduledTask] release delivery contracts failed:', error);
@@ -153,6 +273,117 @@ export async function finalizeScheduledTaskPageDelivery({
  return readyPaths;
 }

+export async function awaitScheduledTaskPageDelivery({
+  pool,
+  userId,
+  sessionId,
+  messages,
+  publishDir,
+  deliveryText = '',
+  tkmindProxy = null,
+  sessionSnapshotService = null,
+  retryDelaysMs = resolveScheduledTaskDeliveryRetryDelaysMs(),
+  sleepFn = (delayMs) => new Promise((resolve) => {
+    setTimeout(resolve, delayMs);
+  }),
+  logger = console,
+} = {}) {
+  let currentMessages = Array.isArray(messages) ? messages : [];
+  let readyPaths = [];
+  const attempts = [0, ...retryDelaysMs];
+
+  for (let attempt = 0; attempt < attempts.length; attempt += 1) {
+    if (attempt > 0) {
+      await sleepFn(attempts[attempt]);
+      currentMessages = await refreshScheduledTaskMessages({
+        userId,
+        sessionId,
+        tkmindProxy,
+        sessionSnapshotService,
+      });
+    }
+    readyPaths = await finalizeScheduledTaskPageDelivery({
+      pool,
+      userId,
+      sessionId,
+      messages: currentMessages,
+      publishDir,
+      deliveryText,
+      logger,
+    }).catch((error) => {
+      logger.warn?.('[ScheduledTask] finalize page delivery failed:', error);
+      return [];
+    });
+
+    const promisesHtml = deliveryTextPromisesPublicHtml(deliveryText)
+      || collectScheduledTaskPageRelativePaths({
+        messages: currentMessages,
+        publishDir,
+        userId,
+        deliveryText,
+      }).relativePaths.length > 0;
+    if (!promisesHtml || readyPaths.length > 0) {
+      break;
+    }
+    logger.warn?.('[ScheduledTask] page delivery still preparing', {
+      userId,
+      sessionId,
+      attempt: attempt + 1,
+      maxAttempts: attempts.length,
+    });
+  }
+
+  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)) continue;
+    if (
+      await markPageDeliveryContractReady({
+        pool,
+        userId: row.user_id,
+        relativePath,
+      })
+    ) {
+      released.push(relativePath);
+      logger.info?.('[ScheduledTask] reconciled static delivery contract', {
+        userId: row.user_id,
+        relativePath,
+      });
+    }
+  }
+  return released;
+}
+
 export async function executeScheduledTask(task, {
  userAuth,
  tkmindProxy,
@@ -204,30 +435,34 @@ export async function executeScheduledTask(task, {
    { timeoutMs },
  );

-  let messages = [];
-  if (typeof tkmindProxy.fetchSessionConversationForUser === 'function') {
-    messages = await tkmindProxy.fetchSessionConversationForUser(task.userId, sessionId).catch(() => []);
-  } else if (typeof sessionSnapshotService?.get === 'function') {
-    const snapshot = await sessionSnapshotService.get(sessionId).catch(() => null);
-    messages = snapshot?.messages ?? snapshot?.conversation?.messages ?? [];
-  }
+  let messages = await refreshScheduledTaskMessages({
+    userId: task.userId,
+    sessionId,
+    tkmindProxy,
+    sessionSnapshotService,
+  });

  const publishDir = h5Root && task.userId
    ? path.join(h5Root, 'MindSpace', task.userId)
    : null;
-  const readyPaths = await finalizeScheduledTaskPageDelivery({
-    pool,
-    userId: task.userId,
-    sessionId,
-    messages,
-    publishDir,
-    logger,
-  }).catch((error) => {
-    logger.warn?.('[ScheduledTask] finalize page delivery failed:', error);
-    return [];
-  });
-
  let deliveryText = extractScheduledTaskDeliveryText(messages, task);
+  const deliveryResult = publishDir
+    ? await awaitScheduledTaskPageDelivery({
+      pool,
+      userId: task.userId,
+      sessionId,
+      messages,
+      publishDir,
+      deliveryText,
+      tkmindProxy,
+      sessionSnapshotService,
+      logger,
+    })
+    : { messages, readyPaths: [] };
+  messages = deliveryResult.messages;
+  const readyPaths = deliveryResult.readyPaths;
+
+  deliveryText = extractScheduledTaskDeliveryText(messages, task);
  if (
    readyPaths.length > 0
    && !/https?:\/\//i.test(deliveryText)