From 81e63c16d32c7406593ec6cd016c5d782df6ef8a Mon Sep 17 00:00:00 2001 From: john Date: Fri, 31 Jul 2026 08:02:49 +0800 Subject: [PATCH] Add scheduled task automation skill with worker execution pipeline. Introduce scheduled-task-automation for H5 and WeChat, persist tasks in h5_scheduled_tasks, and run due jobs via a dedicated worker that executes agent tasks and delivers results. Co-authored-by: Cursor --- capabilities.mjs | 3 + capabilities.test.mjs | 9 + chat-skills.mjs | 15 + chat-skills.test.mjs | 8 + db.mjs | 31 ++ mindspace-sandbox-mcp.mjs | 95 +++++ package.json | 3 +- schedule-intent.mjs | 7 + schedule-intent.test.mjs | 4 + schedule-time.mjs | 50 +++ scheduled-task-executor.mjs | 131 ++++++ scheduled-task-executor.test.mjs | 39 ++ scheduled-task-intent.mjs | 80 ++++ scheduled-task-intent.test.mjs | 48 +++ scheduled-task-service.mjs | 375 ++++++++++++++++++ scheduled-task-service.test.mjs | 168 ++++++++ scheduled-task-worker.mjs | 130 ++++++ scheduled-task-worker.test.mjs | 125 ++++++ scripts/verify-scheduled-task-automation.mjs | 277 +++++++++++++ server.mjs | 7 + server/portal-domain-services-bootstrap.mjs | 7 + .../portal-integration-services-bootstrap.mjs | 28 +- skills-registry.mjs | 1 + skills-registry.test.mjs | 2 + skills/scheduled-task-automation/SKILL.md | 67 ++++ user-memory-profile.mjs | 11 + wechat/prompts/chat-general.mjs | 17 + 27 files changed, 1736 insertions(+), 2 deletions(-) create mode 100644 scheduled-task-executor.mjs create mode 100644 scheduled-task-executor.test.mjs create mode 100644 scheduled-task-intent.mjs create mode 100644 scheduled-task-intent.test.mjs create mode 100644 scheduled-task-service.mjs create mode 100644 scheduled-task-service.test.mjs create mode 100644 scheduled-task-worker.mjs create mode 100644 scheduled-task-worker.test.mjs create mode 100644 scripts/verify-scheduled-task-automation.mjs create mode 100644 skills/scheduled-task-automation/SKILL.md diff --git a/capabilities.mjs b/capabilities.mjs index eb260d8..a909ac6 100644 --- a/capabilities.mjs +++ b/capabilities.mjs @@ -459,6 +459,9 @@ export function sandboxMcpTools(capabilities) { 'schedule_create_item', 'schedule_create_reminder', 'schedule_list_items', + 'scheduled_task_create', + 'scheduled_task_list', + 'scheduled_task_cancel', ); } return tools; diff --git a/capabilities.test.mjs b/capabilities.test.mjs index 643548c..1fa4ced 100644 --- a/capabilities.test.mjs +++ b/capabilities.test.mjs @@ -374,6 +374,9 @@ test('sandboxMcpTools returns correct tool list based on capabilities', () => { 'schedule_create_item', 'schedule_create_reminder', 'schedule_list_items', + 'scheduled_task_create', + 'scheduled_task_list', + 'scheduled_task_cancel', ]); const withBrowse = { ...base, code_browse: true }; @@ -397,6 +400,9 @@ test('private_data_space alone exposes private data tools through sandbox MCP', 'schedule_create_item', 'schedule_create_reminder', 'schedule_list_items', + 'scheduled_task_create', + 'scheduled_task_list', + 'scheduled_task_cancel', ]); const policy = buildAgentExtensionPolicy(caps, { @@ -428,6 +434,9 @@ test('private_data_space alone exposes private data tools through sandbox MCP', 'schedule_create_item', 'schedule_create_reminder', 'schedule_list_items', + 'scheduled_task_create', + 'scheduled_task_list', + 'scheduled_task_cancel', ]); }); diff --git a/chat-skills.mjs b/chat-skills.mjs index 78c656b..e9885b6 100644 --- a/chat-skills.mjs +++ b/chat-skills.mjs @@ -1,7 +1,10 @@ // Keep browser-safe: do not import user-publish.mjs (uses node:fs/path/url). +import { shouldUseScheduledTaskAutomation } from './scheduled-task-intent.mjs'; + const PUBLISH_SKILL_NAME = 'static-page-publish'; export const PAGE_DATA_COLLECT_SKILL_NAME = 'page-data-collect'; export const EXCEL_ANALYST_SKILL_NAME = 'excel-analyst'; +export const SCHEDULED_TASK_AUTOMATION_SKILL_NAME = 'scheduled-task-automation'; export const AIDER_DEVELOPMENT_SKILL_NAME = 'aider-development'; export const SKILL_ROUTER_V2_ENV = 'TKMIND_SKILL_ROUTER_V2'; @@ -297,6 +300,12 @@ export function buildChatSkillPrompt(promptKey, skillName) { '只有用户明确要求 Word / docx 下载时，才先 `load_skill` → `docx-generate`，生成 `public/*.docx` 并确认文件已存在，再写 HTML 并用同目录相对路径链接该文档。' + '只有用户明确要求长图下载时，才先 `load_skill` → `long-image-download`，调用 `generate_long_image` 生成 `public/*.long.png` 并确认文件已存在，再给长图预览链接和 `?download=long-image` 下载链接。不要在没有明确需求时强制生成伴生文件。' ); + case 'scheduled-task-automation': + return ( + `请使用 ${skillName ?? SCHEDULED_TASK_AUTOMATION_SKILL_NAME} 技能：这是定时自动任务（Scheduled Automation），不是待办提醒。` + + '先澄清缺失的执行时间或 taskSpec，再调用 scheduled_task_create / scheduled_task_list / scheduled_task_cancel。' + + '禁止在当前对话立即执行长任务；工具成功返回后才能确认已设置。一次性任务用 runAtLocal，循环任务用 hour/minute/weekday。我的需求是：' + ); default: return ''; } @@ -407,6 +416,12 @@ function buildManifestSkillPrompt(route) { } function buildLegacyAutoChatSkillPrefix(trimmed, grantedSkills) { + if ( + grantedSkills.includes(SCHEDULED_TASK_AUTOMATION_SKILL_NAME) && + shouldUseScheduledTaskAutomation(trimmed) + ) { + return buildChatSkillPrompt('scheduled-task-automation', SCHEDULED_TASK_AUTOMATION_SKILL_NAME); + } if ( grantedSkills.includes(EXCEL_ANALYST_SKILL_NAME) && isExcelAnalysisIntent(trimmed) diff --git a/chat-skills.test.mjs b/chat-skills.test.mjs index 1b30068..e5a8617 100644 --- a/chat-skills.test.mjs +++ b/chat-skills.test.mjs @@ -142,6 +142,14 @@ test('generate-page selection preserves an existing news-page request for the ne assert.equal(mergeChatSkillPromptWithInput(prompt, task), `${prompt}${task}`); }); +test('buildAutoChatSkillPrefix routes scheduled automation when skill is granted', () => { + assert.match( + buildAutoChatSkillPrefix('每天6点帮我做今日新闻页面', ['scheduled-task-automation']), + /scheduled-task-automation/, + ); + assert.equal(buildAutoChatSkillPrefix('每天6点帮我做今日新闻页面', []), ''); +}); + test('buildAutoChatSkillPrefix enables web for news-like queries', () => { assert.match( buildAutoChatSkillPrefix('帮我搜索今天热点新闻', ['web', 'search']), diff --git a/db.mjs b/db.mjs index d1e430b..603d709 100644 --- a/db.mjs +++ b/db.mjs @@ -772,6 +772,37 @@ export async function migrateSchema(pool) { ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci `); + await pool.query(` + CREATE TABLE IF NOT EXISTS h5_scheduled_tasks ( + id CHAR(36) PRIMARY KEY, + user_id CHAR(36) NOT NULL, + title VARCHAR(200) NOT NULL, + task_spec TEXT NOT NULL, + recurrence ENUM('once', 'daily', 'weekly') NOT NULL DEFAULT 'daily', + hour TINYINT UNSIGNED NULL, + minute TINYINT UNSIGNED NOT NULL DEFAULT 0, + weekday TINYINT UNSIGNED NULL, + timezone VARCHAR(64) NOT NULL DEFAULT 'Asia/Shanghai', + next_run_at BIGINT NOT NULL, + last_run_at BIGINT NULL, + notify_channel ENUM('wechat', 'web', 'both') NOT NULL DEFAULT 'both', + status ENUM('active', 'locked', 'running', 'completed', 'failed', 'cancelled') NOT NULL DEFAULT 'active', + attempts INT NOT NULL DEFAULT 0, + locked_until BIGINT NULL, + last_error VARCHAR(500) NULL, + last_result_json JSON NULL, + source_channel ENUM('h5', 'wechat', 'agent', 'api') NOT NULL DEFAULT 'agent', + source_session_id VARCHAR(128) NULL, + source_message_id VARCHAR(128) NULL, + source_text TEXT NULL, + created_at BIGINT NOT NULL, + updated_at BIGINT NOT NULL, + KEY idx_scheduled_task_due (status, next_run_at), + KEY idx_scheduled_task_user (user_id, status, next_run_at), + CONSTRAINT fk_scheduled_task_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci + `); + await pool.query(` CREATE TABLE IF NOT EXISTS h5_schedule_delivery_logs ( id CHAR(36) PRIMARY KEY, diff --git a/mindspace-sandbox-mcp.mjs b/mindspace-sandbox-mcp.mjs index d5be57d..5e3b388 100644 --- a/mindspace-sandbox-mcp.mjs +++ b/mindspace-sandbox-mcp.mjs @@ -16,6 +16,7 @@ import { fileURLToPath } from 'node:url'; import { execFileSync } from 'node:child_process'; import mysql from 'mysql2/promise'; import { createScheduleService } from './schedule-service.mjs'; +import { createScheduledTaskService } from './scheduled-task-service.mjs'; import { resolveScheduleTimestamp } from './schedule-time.mjs'; import { shouldAutoCreateReminderAtStart } from './schedule-service.mjs'; import { renderLongImage } from './mindspace-long-image.mjs'; @@ -484,6 +485,7 @@ const ALL_TOOLS = [ let quotaPool = null; let scheduleService = null; +let scheduledTaskService = null; let userDataSpaceService = null; function isQuotaSyncConfigured() { @@ -529,6 +531,20 @@ function getScheduleService() { return scheduleService; } +function getScheduledTaskService() { + if (!PRIVATE_DATA_USER_ID) throw new Error('当前会话没有用户上下文，不能操作定时任务'); + const pool = getQuotaPool(); + if (!pool || !isScheduleConfigured()) { + throw new Error('当前环境未配置待办数据库连接'); + } + if (!scheduledTaskService) { + scheduledTaskService = createScheduledTaskService(pool, { + defaultTimezone: process.env.H5_DEFAULT_TIMEZONE || 'Asia/Shanghai', + }); + } + return scheduledTaskService; +} + if (isScheduleConfigured()) { ALL_TOOLS.push( { @@ -591,6 +607,50 @@ if (isScheduleConfigured()) { }, }, }, + { + name: 'scheduled_task_create', + description: + '为当前用户创建定时自动任务（Scheduled Automation）。到点由系统自动执行 taskSpec 并推送结果；不用于纯待办/提醒。', + inputSchema: { + type: 'object', + properties: { + title: { type: 'string', description: '任务摘要标题，可选' }, + taskSpec: { type: 'string', description: '到时具体执行什么（必填）' }, + recurrence: { type: 'string', description: 'once / daily / weekly，默认 daily' }, + runAtLocal: { type: 'string', description: '一次性任务执行时间 YYYY-MM-DD HH:mm' }, + hour: { type: 'number', description: 'daily/weekly 的小时 0-23' }, + minute: { type: 'number', description: 'daily/weekly 的分钟 0-59，默认 0' }, + weekday: { type: 'number', description: 'weekly 的星期 0=周日 … 6=周六' }, + timezone: { type: 'string', description: '时区，可选' }, + notifyChannel: { type: 'string', description: 'wechat / web / both，默认 both' }, + sourceMessageId: { type: 'string', description: '服务号消息 ID；有值时必须原样传入' }, + sourceText: { type: 'string', description: '原始用户文本，可选' }, + }, + required: ['taskSpec'], + }, + }, + { + name: 'scheduled_task_list', + description: '查询当前用户的定时自动任务列表。仅返回当前用户数据。', + inputSchema: { + type: 'object', + properties: { + status: { type: 'string', description: '任务状态，默认 active，可选 all' }, + limit: { type: 'number', description: '返回数量，默认 20，可选' }, + }, + }, + }, + { + name: 'scheduled_task_cancel', + description: '取消当前用户的一个定时自动任务。需提供 taskId 或 titleMatch。', + inputSchema: { + type: 'object', + properties: { + taskId: { type: 'string', description: '任务 ID，优先使用' }, + titleMatch: { type: 'string', description: '按标题模糊匹配最近一条 active 任务' }, + }, + }, + }, ); } @@ -1155,6 +1215,41 @@ async function callTool(name, args) { }); return [{ type: 'text', text: JSON.stringify(items, null, 2) }]; } + case 'scheduled_task_create': { + const timezone = args.timezone ?? process.env.H5_DEFAULT_TIMEZONE ?? 'Asia/Shanghai'; + const task = await getScheduledTaskService().createTask({ + userId: PRIVATE_DATA_USER_ID, + title: args.title ?? null, + taskSpec: args.taskSpec, + recurrence: args.recurrence ?? 'daily', + runAtLocal: args.runAtLocal ?? null, + hour: args.hour ?? null, + minute: args.minute ?? 0, + weekday: args.weekday ?? null, + timezone, + notifyChannel: args.notifyChannel ?? 'both', + sourceChannel: 'agent', + sourceMessageId: args.sourceMessageId ?? null, + sourceText: args.sourceText ?? null, + }); + return [{ type: 'text', text: JSON.stringify(task, null, 2) }]; + } + case 'scheduled_task_list': { + const tasks = await getScheduledTaskService().listTasks({ + userId: PRIVATE_DATA_USER_ID, + status: args.status ?? 'active', + limit: args.limit ?? 20, + }); + return [{ type: 'text', text: JSON.stringify(tasks, null, 2) }]; + } + case 'scheduled_task_cancel': { + const task = await getScheduledTaskService().cancelTask({ + userId: PRIVATE_DATA_USER_ID, + taskId: args.taskId ?? null, + titleMatch: args.titleMatch ?? null, + }); + return [{ type: 'text', text: JSON.stringify(task, null, 2) }]; + } default: throw new Error(`未知工具：${name}`); } diff --git a/package.json b/package.json index 1e163a9..0ffe2a1 100644 --- a/package.json +++ b/package.json @@ -78,7 +78,7 @@ "ci:page-data-dev-loop-smoke": "node scripts/ci-page-data-dev-loop-smoke.mjs", "migrate:agent-code-run-config": "node scripts/migrate-agent-code-run-config-from-env.mjs", "test:scenario:john4-diet": "node scripts/run-scenario-test.mjs --scenario john4-children-hobby-diet-update", - "test": "node --test api-core-retry.test.mjs auth.test.mjs asr-proxy.test.mjs billing.test.mjs billing-token-state.test.mjs billing-recharge.test.mjs wechat-pay.test.mjs wechat-oauth.test.mjs wechat-mp.test.mjs wechat-media.test.mjs wechat/image-generation-policy.test.mjs wechat/verify/generated-thumbnail.test.mjs schedule-intent.test.mjs schedule-reminder-worker.test.mjs capabilities.test.mjs policies.test.mjs server/portal-api-auth-middleware.test.mjs server/portal-config-routes.test.mjs server/portal-plaza-discovery-routes.test.mjs server/portal-runtime-routes.test.mjs server/portal-gateway-services-bootstrap.test.mjs chat-skills.test.mjs chat-intent-router.test.mjs chat-finish-sync.test.mjs chat-agent-run-gate.test.mjs conversation-display.test.mjs user-publish.test.mjs user-memory-profile.test.mjs skills-registry.test.mjs skill-runtime-policy.test.mjs excel-analyst.test.mjs agent-run-gateway.test.mjs agent-run-routes.test.mjs session-broker.test.mjs sse-event-taxonomy.test.mjs goosed-proxy-boundary.test.mjs agent-run-stream.test.mjs mindspace-h5-html-finish-guard.test.mjs admin-routes.test.mjs services/orchestrator/admin-config.test.mjs services/orchestrator/contracts.test.mjs services/orchestrator/checkpoint.test.mjs services/orchestrator/runtime.test.mjs services/orchestrator/app.test.mjs services/orchestrator/server.test.mjs services/orchestrator/shadow-dispatcher.test.mjs services/orchestrator/shadow-observer.test.mjs services/orchestrator/observability.test.mjs services/orchestrator/executor-gateway.test.mjs services/orchestrator/executor-job-store.test.mjs image-make-admin-config.test.mjs asset-gateway.test.mjs image-make-client.test.mjs mindspace-image-generation.test.mjs mindspace-image-generation-routes.test.mjs mindspace-image-review.test.mjs mindspace-run-public-html-scope.test.mjs direct-chat-service.test.mjs tool-gateway.test.mjs mindspace.test.mjs mindspace-scan.test.mjs mindspace-assets.test.mjs mindspace-local-runtime-services.test.mjs mindspace-local-server-adapter.test.mjs mindspace-public-asset-token.test.mjs mindspace-remote-server-adapter.test.mjs mindspace-server-adapter.test.mjs mindspace-pages.test.mjs mindspace-page-sync-service.test.mjs public-site-bases.test.mjs mindspace-html-download-links.test.mjs mindspace-long-image.test.mjs mindspace-page-purge.test.mjs mindspace-public-delivery.test.mjs mindspace-public-page-context.test.mjs mindspace-published-page-csp.test.mjs mindspace-published-script-localize.test.mjs agent-run-deliverable-check.test.mjs mindspace-public-route.test.mjs mindspace-publications.test.mjs mindspace-public-links.test.mjs mindspace-chat-save.test.mjs mindspace-chat-save-service.test.mjs mindspace-chat-docx-package.test.mjs mindspace-public-finish-sync.test.mjs mindspace-wechat-html-delivery.test.mjs mindspace-chat-context.test.mjs mindspace-canonical-url.test.mjs mindspace-conversation-package.test.mjs mindspace-conversation-package-artifact-service.test.mjs mindspace-conversation-package-audit.test.mjs mindspace-conversation-package-backfill.test.mjs mindspace-conversation-package-public-html.test.mjs mindspace-conversation-package-verify.test.mjs mindspace-conversation-package-registry.test.mjs mindspace-conversation-package-routes.test.mjs mindspace-conversation-package-store.test.mjs mindspace-conversation-schema.test.mjs mindspace-runtime-config.test.mjs mindspace-config.test.mjs mindspace-analytics.test.mjs mindspace-rybbit.test.mjs mindspace-service.test.mjs mindspace-storage-adapter.test.mjs mindspace-content-scan.test.mjs mindspace-html-localize.test.mjs mindspace-visual-editor.test.mjs mindspace-cleanup.test.mjs mindspace-thumbnails.test.mjs mindspace-workspace-thumbnails.test.mjs mindspace-workspace-sync.test.mjs mindspace-workspace-publication-delivery-service.test.mjs mindspace-workspace-tool-service.test.mjs mindspace-mcp-scoped-token.test.mjs mindspace-asset-preview.test.mjs mindspace-agent-jobs.test.mjs mindspace-agent-runner.test.mjs mindspace-sandbox-mcp.test.mjs mindspace-userdata-postgres.test.mjs postgres-user-data-space-service.test.mjs user-data-space-service.test.mjs page-data-routes.test.mjs page-access-policy.test.mjs page-access-visitor.test.mjs page-data-public-service.test.mjs page-data-integration.test.mjs page-data-log-store.test.mjs page-data-ops.test.mjs page-data-session-store.test.mjs page-data-browser-client.test.mjs page-data-policy-index.test.mjs message-stream.test.mjs mindspace-service/mindspace-rpc-server.test.mjs plaza-posts.test.mjs plaza-interactions.test.mjs plaza-algorithm.test.mjs plaza-seo.test.mjs plaza-ops.test.mjs user-auth.test.mjs llm-providers.test.mjs admin-guard.test.mjs user-feedback.test.mjs memory-v2.test.mjs memory-v2-admin-config.test.mjs memory-v2-lifecycle.test.mjs memory-v2-adapter-scaffold.test.mjs memory-v2-backend-contract.test.mjs memory-v2-health.test.mjs memory-v2-runtime.test.mjs memory-v2-plugin-backends.test.mjs memory-v2-pgvector.test.mjs memory-v2-pgvector-schema.test.mjs memory-v2-pgvector-backfill.test.mjs memory-v2-pgvector-smoke.test.mjs memory-v2-qdrant.test.mjs memory-v2-weaviate.test.mjs memory-v2-mem0.test.mjs memory-v2-letta.test.mjs memory-v2-external-adapters.test.mjs scripts/embed-memory-v2-local-hash.test.mjs scripts/check-memory-v2-app-canary.test.mjs scripts/check-memory-v2-config-gaps.test.mjs scripts/check-memory-v2-contracts.test.mjs scripts/check-memory-v2-health.test.mjs scripts/check-memory-v2-session-flow.test.mjs scripts/check-memory-v2-stack.test.mjs scripts/setup-memory-v2-pgvector-schema.test.mjs scripts/backfill-memory-v2-pgvector.test.mjs scripts/scaffold-memory-v2-backend.test.mjs scripts/smoke-memory-v2-pgvector.test.mjs scripts/smoke-memory-v2-qdrant.test.mjs scripts/smoke-memory-v2-external.test.mjs scripts/mock-memory-v2-services.test.mjs", + "test": "node --test api-core-retry.test.mjs auth.test.mjs asr-proxy.test.mjs billing.test.mjs billing-token-state.test.mjs billing-recharge.test.mjs wechat-pay.test.mjs wechat-oauth.test.mjs wechat-mp.test.mjs wechat-media.test.mjs wechat/image-generation-policy.test.mjs wechat/verify/generated-thumbnail.test.mjs schedule-intent.test.mjs schedule-reminder-worker.test.mjs scheduled-task-intent.test.mjs scheduled-task-service.test.mjs scheduled-task-executor.test.mjs scheduled-task-worker.test.mjs capabilities.test.mjs policies.test.mjs server/portal-api-auth-middleware.test.mjs server/portal-config-routes.test.mjs server/portal-plaza-discovery-routes.test.mjs server/portal-runtime-routes.test.mjs server/portal-gateway-services-bootstrap.test.mjs chat-skills.test.mjs chat-intent-router.test.mjs chat-finish-sync.test.mjs chat-agent-run-gate.test.mjs conversation-display.test.mjs user-publish.test.mjs user-memory-profile.test.mjs skills-registry.test.mjs skill-runtime-policy.test.mjs excel-analyst.test.mjs agent-run-gateway.test.mjs agent-run-routes.test.mjs session-broker.test.mjs sse-event-taxonomy.test.mjs goosed-proxy-boundary.test.mjs agent-run-stream.test.mjs mindspace-h5-html-finish-guard.test.mjs admin-routes.test.mjs services/orchestrator/admin-config.test.mjs services/orchestrator/contracts.test.mjs services/orchestrator/checkpoint.test.mjs services/orchestrator/runtime.test.mjs services/orchestrator/app.test.mjs services/orchestrator/server.test.mjs services/orchestrator/shadow-dispatcher.test.mjs services/orchestrator/shadow-observer.test.mjs services/orchestrator/observability.test.mjs services/orchestrator/executor-gateway.test.mjs services/orchestrator/executor-job-store.test.mjs image-make-admin-config.test.mjs asset-gateway.test.mjs image-make-client.test.mjs mindspace-image-generation.test.mjs mindspace-image-generation-routes.test.mjs mindspace-image-review.test.mjs mindspace-run-public-html-scope.test.mjs direct-chat-service.test.mjs tool-gateway.test.mjs mindspace.test.mjs mindspace-scan.test.mjs mindspace-assets.test.mjs mindspace-local-runtime-services.test.mjs mindspace-local-server-adapter.test.mjs mindspace-public-asset-token.test.mjs mindspace-remote-server-adapter.test.mjs mindspace-server-adapter.test.mjs mindspace-pages.test.mjs mindspace-page-sync-service.test.mjs public-site-bases.test.mjs mindspace-html-download-links.test.mjs mindspace-long-image.test.mjs mindspace-page-purge.test.mjs mindspace-public-delivery.test.mjs mindspace-public-page-context.test.mjs mindspace-published-page-csp.test.mjs mindspace-published-script-localize.test.mjs agent-run-deliverable-check.test.mjs mindspace-public-route.test.mjs mindspace-publications.test.mjs mindspace-public-links.test.mjs mindspace-chat-save.test.mjs mindspace-chat-save-service.test.mjs mindspace-chat-docx-package.test.mjs mindspace-public-finish-sync.test.mjs mindspace-wechat-html-delivery.test.mjs mindspace-chat-context.test.mjs mindspace-canonical-url.test.mjs mindspace-conversation-package.test.mjs mindspace-conversation-package-artifact-service.test.mjs mindspace-conversation-package-audit.test.mjs mindspace-conversation-package-backfill.test.mjs mindspace-conversation-package-public-html.test.mjs mindspace-conversation-package-verify.test.mjs mindspace-conversation-package-registry.test.mjs mindspace-conversation-package-routes.test.mjs mindspace-conversation-package-store.test.mjs mindspace-conversation-schema.test.mjs mindspace-runtime-config.test.mjs mindspace-config.test.mjs mindspace-analytics.test.mjs mindspace-rybbit.test.mjs mindspace-service.test.mjs mindspace-storage-adapter.test.mjs mindspace-content-scan.test.mjs mindspace-html-localize.test.mjs mindspace-visual-editor.test.mjs mindspace-cleanup.test.mjs mindspace-thumbnails.test.mjs mindspace-workspace-thumbnails.test.mjs mindspace-workspace-sync.test.mjs mindspace-workspace-publication-delivery-service.test.mjs mindspace-workspace-tool-service.test.mjs mindspace-mcp-scoped-token.test.mjs mindspace-asset-preview.test.mjs mindspace-agent-jobs.test.mjs mindspace-agent-runner.test.mjs mindspace-sandbox-mcp.test.mjs mindspace-userdata-postgres.test.mjs postgres-user-data-space-service.test.mjs user-data-space-service.test.mjs page-data-routes.test.mjs page-access-policy.test.mjs page-access-visitor.test.mjs page-data-public-service.test.mjs page-data-integration.test.mjs page-data-log-store.test.mjs page-data-ops.test.mjs page-data-session-store.test.mjs page-data-browser-client.test.mjs page-data-policy-index.test.mjs message-stream.test.mjs mindspace-service/mindspace-rpc-server.test.mjs plaza-posts.test.mjs plaza-interactions.test.mjs plaza-algorithm.test.mjs plaza-seo.test.mjs plaza-ops.test.mjs user-auth.test.mjs llm-providers.test.mjs admin-guard.test.mjs user-feedback.test.mjs memory-v2.test.mjs memory-v2-admin-config.test.mjs memory-v2-lifecycle.test.mjs memory-v2-adapter-scaffold.test.mjs memory-v2-backend-contract.test.mjs memory-v2-health.test.mjs memory-v2-runtime.test.mjs memory-v2-plugin-backends.test.mjs memory-v2-pgvector.test.mjs memory-v2-pgvector-schema.test.mjs memory-v2-pgvector-backfill.test.mjs memory-v2-pgvector-smoke.test.mjs memory-v2-qdrant.test.mjs memory-v2-weaviate.test.mjs memory-v2-mem0.test.mjs memory-v2-letta.test.mjs memory-v2-external-adapters.test.mjs scripts/embed-memory-v2-local-hash.test.mjs scripts/check-memory-v2-app-canary.test.mjs scripts/check-memory-v2-config-gaps.test.mjs scripts/check-memory-v2-contracts.test.mjs scripts/check-memory-v2-health.test.mjs scripts/check-memory-v2-session-flow.test.mjs scripts/check-memory-v2-stack.test.mjs scripts/setup-memory-v2-pgvector-schema.test.mjs scripts/backfill-memory-v2-pgvector.test.mjs scripts/scaffold-memory-v2-backend.test.mjs scripts/smoke-memory-v2-pgvector.test.mjs scripts/smoke-memory-v2-qdrant.test.mjs scripts/smoke-memory-v2-external.test.mjs scripts/mock-memory-v2-services.test.mjs", "test:episodic-memory": "node --test episodic-memory.test.mjs direct-chat-service.test.mjs chat-intent-router.test.mjs", "test:deep-search": "node --test deep-search.test.mjs mindsearch.test.mjs", "test:image-review": "node --test mindspace-image-review.test.mjs mindspace-image-generation.test.mjs", @@ -100,6 +100,7 @@ "verify:wechat-channel-isolation": "node scripts/check-wechat-channel-isolation.mjs", "verify:goosed-proxy-boundary": "node scripts/check-goosed-proxy-boundary.mjs", "verify:h5-session-patches": "node scripts/verify-h5-session-patches.mjs", + "verify:scheduled-task-automation": "node scripts/verify-scheduled-task-automation.mjs", "test:mindspace-e2e": "node scripts/mindspace-e2e.mjs", "test:mindspace-pages-e2e": "node scripts/mindspace-pages-e2e.mjs", "test:mindspace-publications-e2e": "node scripts/mindspace-publications-e2e.mjs", diff --git a/schedule-intent.mjs b/schedule-intent.mjs index 9967f67..29b4981 100644 --- a/schedule-intent.mjs +++ b/schedule-intent.mjs @@ -140,6 +140,13 @@ export function isScheduleIntent(intent) { export function shouldUseScheduleAssistant(text) { const compact = normalizeText(text); if (!compact) return false; + // Scheduled automation (execute + deliver) uses scheduled-task-automation skill. + if (/(?:定时(?:自动)?任务|scheduled\s*task|scheduled\s*automation)/iu.test(compact)) return false; + if (/(?:每天|每日|每周|定时|到点|届时|自动)/u.test(compact) + && /(?:做|生成|制作|创建|写|执行|跑|更新|整理|汇总|推送)/u.test(compact) + && !/(?:提醒我|待办记录|待办列表|待办清单|当天待办|一天的待办)/u.test(compact)) { + return false; + } if (isScheduleIntent(parseScheduleIntent(text))) return true; const scheduleKeywords = /(提醒|待办|代办|带办|代拜|日程|行程|安排|计划|闹钟)/; const timeKeywords = /(今天|今晚|明天|后天|早上|上午|中午|下午|晚上|\d{1,2}[点:：])/; diff --git a/schedule-intent.test.mjs b/schedule-intent.test.mjs index 0c4d1ec..6c79889 100644 --- a/schedule-intent.test.mjs +++ b/schedule-intent.test.mjs @@ -3,6 +3,10 @@ import test from 'node:test'; import { parseScheduleIntent, shouldUseScheduleAssistant } from './schedule-intent.mjs'; import { nextDailyRunAt } from './schedule-time.mjs'; +test('daily news automation does not route to schedule assistant', () => { + assert.equal(shouldUseScheduleAssistant('每天6点帮我做今日新闻页面'), false); +}); + test('parses daily 7am todo digest request', () => { const intent = parseScheduleIntent('每天早上 7 点给我发一天的待办记录'); assert.equal(intent.action, 'create_daily_todo_digest'); diff --git a/schedule-time.mjs b/schedule-time.mjs index cb6ad46..b2696b7 100644 --- a/schedule-time.mjs +++ b/schedule-time.mjs @@ -113,6 +113,56 @@ export function zonedTimeToEpochMs(parts, timezone = DEFAULT_TIMEZONE) { return guess; } +export function nextWeeklyRunAt({ + weekday, + hour, + minute = 0, + timezone = DEFAULT_TIMEZONE, + now = Date.now(), +} = {}) { + const tz = normalizeTimezone(timezone); + const safeWeekday = Number(weekday); + if (!Number.isInteger(safeWeekday) || safeWeekday < 0 || safeWeekday > 6) { + throw new Error('weekly 任务需要 weekday（0=周日 … 6=周六）'); + } + const todayStart = startOfLocalDay(now, tz); + const todayParts = getLocalParts(todayStart, tz); + const todayWeekday = new Date( + Date.UTC(todayParts.year, todayParts.month - 1, todayParts.day), + ).getUTCDay(); + + const dayOffset = (safeWeekday - todayWeekday + 7) % 7; + let candidateDayStart = addLocalDays(todayStart, dayOffset, tz); + let parts = getLocalParts(candidateDayStart, tz); + let runAt = zonedTimeToEpochMs( + { + year: parts.year, + month: parts.month, + day: parts.day, + hour: Number(hour), + minute: Number(minute), + second: 0, + }, + tz, + ); + if (runAt <= now) { + candidateDayStart = addLocalDays(candidateDayStart, 7, tz); + parts = getLocalParts(candidateDayStart, tz); + runAt = zonedTimeToEpochMs( + { + year: parts.year, + month: parts.month, + day: parts.day, + hour: Number(hour), + minute: Number(minute), + second: 0, + }, + tz, + ); + } + return runAt; +} + export function nextDailyRunAt({ hour, minute = 0, timezone = DEFAULT_TIMEZONE, now = Date.now() }) { const tz = normalizeTimezone(timezone); const todayStart = startOfLocalDay(now, tz); diff --git a/scheduled-task-executor.mjs b/scheduled-task-executor.mjs new file mode 100644 index 0000000..3c538ee --- /dev/null +++ b/scheduled-task-executor.mjs @@ -0,0 +1,131 @@ +import crypto from 'node:crypto'; +import { buildChatSkillPrompt, SCHEDULED_TASK_AUTOMATION_SKILL_NAME } from './chat-skills.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}`, + '执行约束：', + '- 这是系统自动触发的定时任务，请直接完成可交付结果，不要反问用户。', + '- 若需要生成页面，必须先 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 async function executeScheduledTask(task, { + userAuth, + tkmindProxy, + sessionSnapshotService = 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 = []; + if (typeof sessionSnapshotService?.refresh === 'function') { + const refreshed = await sessionSnapshotService.refresh(sessionId).catch(() => null); + messages = refreshed?.messages ?? refreshed?.conversation?.messages ?? []; + } else if (typeof sessionSnapshotService?.get === 'function') { + const snapshot = await sessionSnapshotService.get(sessionId).catch(() => null); + messages = snapshot?.messages ?? snapshot?.conversation?.messages ?? []; + } + + const deliveryText = extractScheduledTaskDeliveryText(messages, task); + return { + sessionId, + requestId, + deliveryText, + messages, + }; +} diff --git a/scheduled-task-executor.test.mjs b/scheduled-task-executor.test.mjs new file mode 100644 index 0000000..1042bec --- /dev/null +++ b/scheduled-task-executor.test.mjs @@ -0,0 +1,39 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { + buildScheduledTaskExecutionPrompt, + extractScheduledTaskDeliveryText, + formatScheduledTaskDeliveryMessage, +} from './scheduled-task-executor.mjs'; + +test('buildScheduledTaskExecutionPrompt includes task spec and automation marker', () => { + const message = buildScheduledTaskExecutionPrompt({ + id: 'task-1', + title: '每日新闻页', + taskSpec: '搜索今日新闻并生成 HTML 页面', + recurrence: 'daily', + timezone: 'Asia/Shanghai', + }); + const text = message.content[0].text; + assert.match(text, /scheduled-task-automation/); + assert.match(text, /每日新闻页/); + assert.match(text, /搜索今日新闻并生成 HTML 页面/); + assert.match(text, /Scheduled Automation/); +}); + +test('extractScheduledTaskDeliveryText reads last assistant message', () => { + const text = extractScheduledTaskDeliveryText([ + { role: 'user', content: [{ type: 'text', text: 'hi' }] }, + { role: 'assistant', content: [{ type: 'text', text: '今日新闻页：https://example.com/news.html' }] }, + ], { title: '每日新闻页' }); + assert.match(text, /https:\/\/example.com\/news.html/); +}); + +test('formatScheduledTaskDeliveryMessage wraps delivery body', () => { + const text = formatScheduledTaskDeliveryMessage( + { title: '每日新闻页' }, + '页面已生成：https://example.com/news.html', + ); + assert.match(text, /定时任务完成：每日新闻页/); + assert.match(text, /页面已生成/); +}); diff --git a/scheduled-task-intent.mjs b/scheduled-task-intent.mjs new file mode 100644 index 0000000..2077301 --- /dev/null +++ b/scheduled-task-intent.mjs @@ -0,0 +1,80 @@ +function normalizeText(text) { + return String(text ?? '').replace(/\s+/g, '').trim(); +} + +const EXECUTE_VERBS = /(?:做|生成|制作|创建|写|执行|跑|更新|整理|汇总|推送)/u; +const SCHEDULE_MARKERS = /(?:定时(?:自动)?任务|scheduled\s*task|scheduled\s*automation|cron\s*job|recurring\s*task)/iu; +const RECURRENCE_MARKERS = /(?:每天|每日|每周|定时|到点|届时|自动)/u; + +function wantsScheduledTaskAutomation(compact) { + if (SCHEDULE_MARKERS.test(compact)) return true; + if (!RECURRENCE_MARKERS.test(compact)) return false; + if (!EXECUTE_VERBS.test(compact)) return false; + // Pure reminders/digests stay on schedule-assistant or wechat schedule handler. + if (/(?:提醒我|闹钟|待办记录|待办列表|待办清单|当天待办|一天的待办)/u.test(compact)) return false; + if (/(?:待办|代办|带办|代拜).{0,8}(?:记录|列表|清单|摘要|汇总)/u.test(compact)) return false; + return true; +} + +export function shouldUseScheduledTaskAutomation(text) { + const compact = normalizeText(text); + if (!compact) return false; + return wantsScheduledTaskAutomation(compact); +} + +export function parseScheduledTaskIntent(text) { + const compact = normalizeText(text); + if (!compact) return { action: 'none' }; + + if (!wantsScheduledTaskAutomation(compact)) { + return { action: 'none' }; + } + + const wantsCancel = /(?:取消|停止|关闭|删除).{0,12}(?:定时|自动)/u.test(compact) + || /(?:cancel|stop|disable).{0,12}(?:scheduled|automation|task)/iu.test(compact); + if (wantsCancel) { + return { action: 'cancel_scheduled_task' }; + } + + const wantsList = /(?:查看|看看|列出|我的).{0,12}(?:定时|自动).{0,12}(?:任务|自动化)/u.test(compact) + || /(?:list|show).{0,12}(?:scheduled|automation).{0,12}tasks/iu.test(compact); + if (wantsList) { + return { action: 'list_scheduled_tasks' }; + } + + const recurrence = /(?:一次|单次|仅一次|once)/iu.test(compact) + ? 'once' + : /(?:每周|weekly)/iu.test(compact) + ? 'weekly' + : /(?:每天|每日|daily)/u.test(compact) + ? 'daily' + : null; + + const hasTaskSpec = EXECUTE_VERBS.test(compact) + && !/^(?:帮我)?(?:设|设置|创建|添加)(?:一个|个)?定时(?:自动)?任务/u.test(String(text ?? '').trim()); + + const needsClarification = []; + if (!recurrence && !/(?:明天|后天|今天|今晚|\d{1,2}[点:：]|[零一二两三四五六七八九十]{1,3}点)/u.test(compact)) { + needsClarification.push('schedule'); + } + if (!hasTaskSpec) { + needsClarification.push('task_spec'); + } + + if (needsClarification.length > 0) { + return { + action: 'create_scheduled_task', + needsClarification, + recurrence: recurrence ?? 'daily', + }; + } + + return { + action: 'create_scheduled_task', + recurrence: recurrence ?? 'daily', + }; +} + +export function isScheduledTaskIntent(intent) { + return intent?.action && intent.action !== 'none'; +} diff --git a/scheduled-task-intent.test.mjs b/scheduled-task-intent.test.mjs new file mode 100644 index 0000000..e7ad5b8 --- /dev/null +++ b/scheduled-task-intent.test.mjs @@ -0,0 +1,48 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { + isScheduledTaskIntent, + parseScheduledTaskIntent, + shouldUseScheduledTaskAutomation, +} from './scheduled-task-intent.mjs'; +import { shouldUseScheduleAssistant } from './schedule-intent.mjs'; + +test('detects daily news page automation intent', () => { + assert.equal(shouldUseScheduledTaskAutomation('每天6点帮我做今日新闻页面'), true); + const intent = parseScheduledTaskIntent('每天6点帮我做今日新闻页面'); + assert.equal(intent.action, 'create_scheduled_task'); + assert.equal(intent.recurrence, 'daily'); +}); + +test('detects explicit scheduled task phrase', () => { + assert.equal(shouldUseScheduledTaskAutomation('帮我设一个定时自动任务'), true); + const intent = parseScheduledTaskIntent('帮我设一个定时自动任务'); + assert.equal(intent.action, 'create_scheduled_task'); + assert.ok(intent.needsClarification?.includes('schedule')); + assert.ok(intent.needsClarification?.includes('task_spec')); +}); + +test('does not capture daily todo digest', () => { + assert.equal(shouldUseScheduledTaskAutomation('每天早上7点把当天待办发给我'), false); + assert.equal(parseScheduledTaskIntent('每天早上7点把当天待办发给我').action, 'none'); +}); + +test('does not capture plain reminders', () => { + assert.equal(shouldUseScheduledTaskAutomation('明天下午三点提醒我开会'), false); + assert.equal(shouldUseScheduleAssistant('明天下午三点提醒我开会'), true); +}); + +test('scheduled automation takes precedence over schedule assistant keyword overlap', () => { + assert.equal(shouldUseScheduledTaskAutomation('每天6点定时任务做新闻页'), true); + assert.equal(shouldUseScheduleAssistant('每天6点定时任务做新闻页'), false); +}); + +test('parses cancel and list intents', () => { + assert.equal(parseScheduledTaskIntent('取消我的定时任务').action, 'cancel_scheduled_task'); + assert.equal(parseScheduledTaskIntent('看看我的定时自动任务').action, 'list_scheduled_tasks'); +}); + +test('isScheduledTaskIntent excludes none', () => { + assert.equal(isScheduledTaskIntent({ action: 'create_scheduled_task' }), true); + assert.equal(isScheduledTaskIntent({ action: 'none' }), false); +}); diff --git a/scheduled-task-service.mjs b/scheduled-task-service.mjs new file mode 100644 index 0000000..f327d48 --- /dev/null +++ b/scheduled-task-service.mjs @@ -0,0 +1,375 @@ +import crypto from 'node:crypto'; +import { + nextDailyRunAt, + nextWeeklyRunAt, + normalizeTimezone, + parseLocalDateTimeString, +} from './schedule-time.mjs'; + +const DEFAULT_TIMEZONE = 'Asia/Shanghai'; +const VALID_RECURRENCES = new Set(['once', 'daily', 'weekly']); +const VALID_NOTIFY_CHANNELS = new Set(['wechat', 'web', 'both']); + +function nowMs() { + return Date.now(); +} + +function rowToTask(row) { + if (!row) return null; + return { + id: row.id, + userId: row.user_id, + title: row.title, + taskSpec: row.task_spec, + recurrence: row.recurrence, + hour: row.hour == null ? null : Number(row.hour), + minute: Number(row.minute ?? 0), + weekday: row.weekday == null ? null : Number(row.weekday), + timezone: row.timezone || DEFAULT_TIMEZONE, + nextRunAt: Number(row.next_run_at), + lastRunAt: row.last_run_at == null ? null : Number(row.last_run_at), + notifyChannel: row.notify_channel || 'both', + status: row.status, + attempts: Number(row.attempts ?? 0), + lastError: row.last_error ?? null, + lastResult: parseJsonColumn(row.last_result_json), + sourceChannel: row.source_channel ?? 'agent', + sourceSessionId: row.source_session_id ?? null, + sourceMessageId: row.source_message_id ?? null, + sourceText: row.source_text ?? null, + createdAt: Number(row.created_at), + updatedAt: Number(row.updated_at), + }; +} + +function parseJsonColumn(value) { + if (value == null || value === '') return null; + if (typeof value === 'string') { + try { return JSON.parse(value); } catch { return null; } + } + if (typeof value === 'object') return value; + return null; +} + +export function computeScheduledTaskNextRunAt({ + recurrence, + runAtLocal = null, + hour = null, + minute = 0, + weekday = null, + timezone = DEFAULT_TIMEZONE, + now = Date.now(), +} = {}) { + const safeRecurrence = String(recurrence ?? '').trim(); + if (!VALID_RECURRENCES.has(safeRecurrence)) { + throw new Error('recurrence 无效，仅支持 once / daily / weekly'); + } + const tz = normalizeTimezone(timezone); + if (safeRecurrence === 'once') { + const runAt = parseLocalDateTimeString(runAtLocal, tz); + if (runAt == null) throw new Error('一次性任务需要 runAtLocal（YYYY-MM-DD HH:mm）'); + if (runAt <= now) throw new Error('一次性任务的执行时间必须在未来'); + return runAt; + } + const safeHour = Number(hour); + const safeMinute = Number(minute ?? 0); + if (!Number.isInteger(safeHour) || safeHour < 0 || safeHour > 23) { + throw new Error('daily/weekly 任务需要有效 hour（0-23）'); + } + if (!Number.isInteger(safeMinute) || safeMinute < 0 || safeMinute > 59) { + throw new Error('minute 无效（0-59）'); + } + if (safeRecurrence === 'weekly') { + return nextWeeklyRunAt({ + weekday, + hour: safeHour, + minute: safeMinute, + timezone: tz, + now, + }); + } + return nextDailyRunAt({ + hour: safeHour, + minute: safeMinute, + timezone: tz, + now, + }); +} + +export function createScheduledTaskService(pool, { defaultTimezone = DEFAULT_TIMEZONE, clock = { now: nowMs } } = {}) { + if (!pool) throw new Error('缺少数据库连接'); + + const createTask = async ({ + userId, + title, + taskSpec, + recurrence = 'daily', + runAtLocal = null, + hour = null, + minute = 0, + weekday = null, + timezone = defaultTimezone, + notifyChannel = 'both', + sourceChannel = 'agent', + sourceSessionId = null, + sourceMessageId = null, + sourceText = null, + }) => { + if (!userId) throw new Error('缺少用户'); + const safeTitle = String(title ?? '').trim(); + const safeTaskSpec = String(taskSpec ?? '').trim(); + if (!safeTaskSpec) throw new Error('缺少 taskSpec（执行内容）'); + const safeRecurrence = String(recurrence ?? 'daily').trim(); + if (!VALID_RECURRENCES.has(safeRecurrence)) { + throw new Error('recurrence 无效，仅支持 once / daily / weekly'); + } + const safeNotifyChannel = String(notifyChannel ?? 'both').trim(); + if (!VALID_NOTIFY_CHANNELS.has(safeNotifyChannel)) { + throw new Error('notifyChannel 无效，仅支持 wechat / web / both'); + } + const tz = normalizeTimezone(timezone); + const nextRunAt = computeScheduledTaskNextRunAt({ + recurrence: safeRecurrence, + runAtLocal, + hour, + minute, + weekday, + timezone: tz, + now: clock.now(), + }); + const id = crypto.randomUUID(); + const ts = clock.now(); + await pool.query( + `INSERT INTO h5_scheduled_tasks + (id, user_id, title, task_spec, recurrence, hour, minute, weekday, timezone, + next_run_at, notify_channel, status, source_channel, source_session_id, + source_message_id, source_text, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'active', ?, ?, ?, ?, ?, ?)`, + [ + id, + userId, + safeTitle || safeTaskSpec.slice(0, 80), + safeTaskSpec, + safeRecurrence, + hour == null ? null : Number(hour), + Number(minute ?? 0), + weekday == null ? null : Number(weekday), + tz, + nextRunAt, + safeNotifyChannel, + sourceChannel, + sourceSessionId, + sourceMessageId, + sourceText, + ts, + ts, + ], + ); + const [rows] = await pool.query( + `SELECT * FROM h5_scheduled_tasks WHERE id = ? LIMIT 1`, + [id], + ); + return rowToTask(rows[0]); + }; + + const listTasks = async ({ + userId, + status = 'active', + limit = 20, + } = {}) => { + if (!userId) throw new Error('缺少用户'); + const clauses = ['user_id = ?']; + const params = [userId]; + if (status && status !== 'all') { + const statuses = Array.isArray(status) ? status : [status]; + clauses.push(`status IN (${statuses.map(() => '?').join(', ')})`); + params.push(...statuses); + } + params.push(Math.max(1, Math.min(200, Number(limit) || 20))); + const [rows] = await pool.query( + `SELECT * + FROM h5_scheduled_tasks + WHERE ${clauses.join(' AND ')} + ORDER BY next_run_at ASC, created_at DESC + LIMIT ?`, + params, + ); + return rows.map(rowToTask); + }; + + const cancelTask = async ({ + userId, + taskId = null, + titleMatch = null, + } = {}) => { + if (!userId) throw new Error('缺少用户'); + const ts = clock.now(); + if (taskId) { + const [result] = await pool.query( + `UPDATE h5_scheduled_tasks + SET status = 'cancelled', updated_at = ? + WHERE id = ? AND user_id = ? AND status IN ('active', 'locked', 'failed')`, + [ts, taskId, userId], + ); + if (Number(result?.affectedRows ?? 0) !== 1) { + throw new Error('未找到可取消的定时任务'); + } + const [rows] = await pool.query( + `SELECT * FROM h5_scheduled_tasks WHERE id = ? LIMIT 1`, + [taskId], + ); + return rowToTask(rows[0]); + } + const safeTitleMatch = String(titleMatch ?? '').trim(); + if (!safeTitleMatch) throw new Error('取消任务需要 taskId 或 titleMatch'); + const [rows] = await pool.query( + `SELECT * + FROM h5_scheduled_tasks + WHERE user_id = ? + AND status IN ('active', 'locked', 'failed') + AND title LIKE ? + ORDER BY created_at DESC + LIMIT 1`, + [userId, `%${safeTitleMatch}%`], + ); + const target = rows[0]; + if (!target) throw new Error('未找到可取消的定时任务'); + await pool.query( + `UPDATE h5_scheduled_tasks + SET status = 'cancelled', updated_at = ? + WHERE id = ?`, + [ts, target.id], + ); + return rowToTask({ ...target, status: 'cancelled', updated_at: ts }); + }; + + const listDueTasks = async ({ now = clock.now(), limit = 50 } = {}) => { + const [rows] = await pool.query( + `SELECT * + FROM h5_scheduled_tasks + WHERE next_run_at <= ? + AND (status = 'active' OR (status = 'locked' AND locked_until IS NOT NULL AND locked_until <= ?)) + ORDER BY next_run_at ASC + LIMIT ?`, + [now, now, Math.max(1, Math.min(200, Number(limit) || 50))], + ); + return rows.map(rowToTask); + }; + + const lockTask = async (id, { now = clock.now(), lockMs = 300_000 } = {}) => { + const lockedUntil = now + lockMs; + const [result] = await pool.query( + `UPDATE h5_scheduled_tasks + SET status = 'locked', locked_until = ?, attempts = attempts + 1, updated_at = ? + WHERE id = ? AND (status = 'active' OR (status = 'locked' AND locked_until IS NOT NULL AND locked_until <= ?))`, + [lockedUntil, now, id, now], + ); + if (Number(result?.affectedRows ?? 0) !== 1) return null; + const [rows] = await pool.query( + `SELECT * FROM h5_scheduled_tasks WHERE id = ? LIMIT 1`, + [id], + ); + return rowToTask(rows[0]); + }; + + const markTaskRunning = async (task, { now = clock.now() } = {}) => { + await pool.query( + `UPDATE h5_scheduled_tasks + SET status = 'running', updated_at = ? + WHERE id = ?`, + [now, task.id], + ); + return { ...task, status: 'running' }; + }; + + const markTaskSucceeded = async (task, { + now = clock.now(), + result = null, + deliveryText = null, + sessionId = null, + requestId = null, + } = {}) => { + const lastResult = { + ...(result && typeof result === 'object' ? result : {}), + deliveryText: deliveryText ?? null, + sessionId: sessionId ?? null, + requestId: requestId ?? null, + finishedAt: now, + }; + if (task.recurrence === 'once') { + await pool.query( + `UPDATE h5_scheduled_tasks + SET status = 'completed', last_run_at = ?, locked_until = NULL, + last_error = NULL, last_result_json = ?, updated_at = ? + WHERE id = ?`, + [now, JSON.stringify(lastResult), now, task.id], + ); + return { + ...task, + status: 'completed', + lastRunAt: now, + lastResult, + }; + } + const nextRunAt = computeScheduledTaskNextRunAt({ + recurrence: task.recurrence, + hour: task.hour, + minute: task.minute, + weekday: task.weekday, + timezone: task.timezone, + now: now + 1000, + }); + await pool.query( + `UPDATE h5_scheduled_tasks + SET status = 'active', next_run_at = ?, last_run_at = ?, locked_until = NULL, + last_error = NULL, last_result_json = ?, updated_at = ? + WHERE id = ?`, + [nextRunAt, now, JSON.stringify(lastResult), now, task.id], + ); + return { + ...task, + status: 'active', + nextRunAt, + lastRunAt: now, + lastResult, + }; + }; + + const markTaskFailed = async (task, error, { + now = clock.now(), + retryMs = 10 * 60 * 1000, + maxAttempts = 5, + } = {}) => { + const attempts = Number(task.attempts ?? 0); + const message = String(error?.message ?? error ?? '执行失败').slice(0, 500); + const terminal = attempts >= maxAttempts; + const nextStatus = terminal ? 'failed' : 'active'; + const nextRunAt = terminal + ? task.nextRunAt + : now + retryMs; + await pool.query( + `UPDATE h5_scheduled_tasks + SET status = ?, next_run_at = ?, locked_until = NULL, last_error = ?, updated_at = ? + WHERE id = ?`, + [nextStatus, nextRunAt, message, now, task.id], + ); + return { + ...task, + status: nextStatus, + nextRunAt, + lastError: message, + }; + }; + + return { + createTask, + listTasks, + cancelTask, + listDueTasks, + lockTask, + markTaskRunning, + markTaskSucceeded, + markTaskFailed, + computeScheduledTaskNextRunAt, + }; +} diff --git a/scheduled-task-service.test.mjs b/scheduled-task-service.test.mjs new file mode 100644 index 0000000..32a5b05 --- /dev/null +++ b/scheduled-task-service.test.mjs @@ -0,0 +1,168 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { createScheduledTaskService, computeScheduledTaskNextRunAt } from './scheduled-task-service.mjs'; +import { nextDailyRunAt } from './schedule-time.mjs'; + +test('computeScheduledTaskNextRunAt supports daily schedule', () => { + const now = Date.parse('2026-07-31T02:00:00+08:00'); + const nextRunAt = computeScheduledTaskNextRunAt({ + recurrence: 'daily', + hour: 6, + minute: 0, + timezone: 'Asia/Shanghai', + now, + }); + assert.equal(nextRunAt, nextDailyRunAt({ hour: 6, minute: 0, timezone: 'Asia/Shanghai', now })); +}); + +test('createTask inserts scheduled automation row', async () => { + const inserts = []; + const service = createScheduledTaskService({ + async query(sql, params) { + if (sql.includes('INSERT INTO h5_scheduled_tasks')) { + inserts.push(params); + return [{ affectedRows: 1 }]; + } + if (sql.includes('SELECT * FROM h5_scheduled_tasks WHERE id = ?')) { + return [[{ + id: params[0], + user_id: 'user-1', + title: '每日新闻页', + task_spec: '搜索今日新闻并生成 HTML 页面', + recurrence: 'daily', + hour: 6, + minute: 0, + weekday: null, + timezone: 'Asia/Shanghai', + next_run_at: 1780000000000, + last_run_at: null, + notify_channel: 'both', + status: 'active', + attempts: 0, + last_error: null, + last_result_json: null, + source_channel: 'agent', + source_session_id: null, + source_message_id: null, + source_text: null, + created_at: 1780000000000, + updated_at: 1780000000000, + }]]; + } + return [[]]; + }, + }, { + clock: { now: () => 1780000000000 }, + }); + + const task = await service.createTask({ + userId: 'user-1', + title: '每日新闻页', + taskSpec: '搜索今日新闻并生成 HTML 页面', + recurrence: 'daily', + hour: 6, + minute: 0, + }); + + assert.equal(task.title, '每日新闻页'); + assert.equal(task.recurrence, 'daily'); + assert.equal(inserts.length, 1); +}); + +test('cancelTask updates status to cancelled', async () => { + const service = createScheduledTaskService({ + async query(sql, params) { + if (sql.includes('UPDATE h5_scheduled_tasks') && sql.includes('cancelled')) { + return [{ affectedRows: 1 }]; + } + if (sql.includes('SELECT * FROM h5_scheduled_tasks WHERE id = ?')) { + return [[{ + id: params[0], + user_id: 'user-1', + title: '每日新闻页', + task_spec: '搜索今日新闻并生成 HTML 页面', + recurrence: 'daily', + hour: 6, + minute: 0, + weekday: null, + timezone: 'Asia/Shanghai', + next_run_at: 1780000000000, + last_run_at: null, + notify_channel: 'both', + status: 'cancelled', + attempts: 0, + last_error: null, + last_result_json: null, + source_channel: 'agent', + source_session_id: null, + source_message_id: null, + source_text: null, + created_at: 1780000000000, + updated_at: 1780000001000, + }]]; + } + return [[]]; + }, + }); + + const task = await service.cancelTask({ userId: 'user-1', taskId: 'task-1' }); + assert.equal(task.status, 'cancelled'); +}); + +test('markTaskSucceeded completes one-shot task', async () => { + const updates = []; + const service = createScheduledTaskService({ + async query(sql, params) { + if (sql.includes('UPDATE h5_scheduled_tasks') && sql.includes("'completed'")) { + updates.push(['completed', params]); + } + return [{ affectedRows: 1 }]; + }, + }, { + clock: { now: () => 1780000001000 }, + }); + + const result = await service.markTaskSucceeded({ + id: 'task-1', + recurrence: 'once', + title: '一次性任务', + nextRunAt: 1780000000000, + }, { + deliveryText: 'done', + sessionId: 'session-1', + requestId: 'req-1', + }); + + assert.equal(result.status, 'completed'); + assert.equal(updates.length, 1); +}); + +test('markTaskSucceeded schedules next run for daily task', async () => { + const updates = []; + const service = createScheduledTaskService({ + async query(sql, params) { + if (sql.includes('UPDATE h5_scheduled_tasks') && sql.includes("'active'")) { + updates.push(params); + } + return [{ affectedRows: 1 }]; + }, + }, { + clock: { now: () => Date.parse('2026-07-31T02:00:00+08:00') }, + }); + + const result = await service.markTaskSucceeded({ + id: 'task-2', + recurrence: 'daily', + hour: 6, + minute: 0, + timezone: 'Asia/Shanghai', + title: '每日任务', + nextRunAt: Date.parse('2026-07-31T06:00:00+08:00'), + }, { + deliveryText: 'done', + }); + + assert.equal(result.status, 'active'); + assert.ok(result.nextRunAt > Date.parse('2026-07-31T02:00:00+08:00')); + assert.equal(updates.length, 1); +}); diff --git a/scheduled-task-worker.mjs b/scheduled-task-worker.mjs new file mode 100644 index 0000000..4482038 --- /dev/null +++ b/scheduled-task-worker.mjs @@ -0,0 +1,130 @@ +import { + executeScheduledTask, + formatScheduledTaskDeliveryMessage, +} from './scheduled-task-executor.mjs'; + +export function startScheduledTaskWorker({ + scheduledTaskService, + scheduleService = null, + executeTask = executeScheduledTask, + userAuth = null, + tkmindProxy = null, + sessionSnapshotService = null, + notificationDispatcher = null, + logger = console, + intervalMs = Number(process.env.H5_SCHEDULED_TASK_SCAN_INTERVAL_MS ?? 30_000), + maxAttempts = Number(process.env.H5_SCHEDULED_TASK_MAX_ATTEMPTS ?? 3), + executionTimeoutMs = Number(process.env.H5_SCHEDULED_TASK_EXECUTION_TIMEOUT_MS ?? 15 * 60 * 1000), + runOnStart = true, +} = {}) { + const sendScheduleNotification = + notificationDispatcher?.sendScheduleNotification?.bind(notificationDispatcher) ?? + null; + const canExecute = + scheduledTaskService + && typeof executeTask === 'function' + && userAuth + && tkmindProxy; + if (!canExecute) { + return { stop() {} }; + } + + let stopped = false; + let running = false; + + const deliverTaskResult = async (task, deliveryText) => { + const text = formatScheduledTaskDeliveryMessage(task, deliveryText); + const notifyChannel = task.notifyChannel ?? 'both'; + if (scheduleService?.createUserNotification && (notifyChannel === 'web' || notifyChannel === 'both')) { + await scheduleService.createUserNotification({ + userId: task.userId, + channel: 'web', + notificationType: 'scheduled_task_result', + title: `定时任务完成：${task.title}`, + body: text, + data: { + taskId: task.id, + recurrence: task.recurrence, + }, + }); + } + if ( + (notifyChannel === 'wechat' || notifyChannel === 'both') + && typeof sendScheduleNotification === 'function' + ) { + await sendScheduleNotification({ userId: task.userId, text }); + } + }; + + const runOnce = async () => { + if (running || stopped) return; + running = true; + try { + const dueTasks = await scheduledTaskService.listDueTasks({ limit: 10 }); + for (const candidate of dueTasks) { + const task = await scheduledTaskService.lockTask(candidate.id); + if (!task) continue; + try { + await scheduledTaskService.markTaskRunning(task); + const result = await executeTask(task, { + userAuth, + tkmindProxy, + sessionSnapshotService, + timeoutMs: executionTimeoutMs, + logger, + }); + await deliverTaskResult(task, result.deliveryText); + await scheduledTaskService.markTaskSucceeded(task, { + result: { + deliveryText: result.deliveryText, + }, + deliveryText: result.deliveryText, + sessionId: result.sessionId, + requestId: result.requestId, + }); + } catch (err) { + logger.warn?.('Scheduled task execution failed:', err); + const failedTask = await scheduledTaskService.markTaskFailed(task, err, { + maxAttempts, + }); + if (failedTask.status === 'failed' && scheduleService?.createUserNotification) { + await scheduleService.createUserNotification({ + userId: task.userId, + channel: 'web', + notificationType: 'scheduled_task_failed', + title: `定时任务失败：${task.title}`, + body: String(failedTask.lastError ?? '执行失败'), + data: { taskId: task.id }, + }).catch(() => {}); + } + if ( + failedTask.status === 'failed' + && (task.notifyChannel === 'wechat' || task.notifyChannel === 'both') + && typeof sendScheduleNotification === 'function' + ) { + await sendScheduleNotification({ + userId: task.userId, + text: `定时任务失败：${task.title}\n${failedTask.lastError ?? '执行失败'}`, + }).catch(() => {}); + } + } + } + } catch (err) { + logger.warn?.('Scheduled task worker failed:', err); + } finally { + running = false; + } + }; + + const timer = setInterval(runOnce, Math.max(1000, Number(intervalMs) || 30_000)); + timer.unref?.(); + if (runOnStart) void runOnce(); + + return { + runOnce, + stop() { + stopped = true; + clearInterval(timer); + }, + }; +} diff --git a/scheduled-task-worker.test.mjs b/scheduled-task-worker.test.mjs new file mode 100644 index 0000000..31bd9d2 --- /dev/null +++ b/scheduled-task-worker.test.mjs @@ -0,0 +1,125 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { startScheduledTaskWorker } from './scheduled-task-worker.mjs'; + +test('scheduled task worker executes due task and completes one-shot task', async () => { + const calls = []; + const notifications = []; + const wechat = []; + const task = { + id: 'task-1', + userId: 'user-1', + title: '每日新闻页', + taskSpec: '做新闻页', + recurrence: 'once', + notifyChannel: 'both', + attempts: 1, + }; + const worker = startScheduledTaskWorker({ + intervalMs: 60_000, + userAuth: { id: 'user-auth' }, + tkmindProxy: { id: 'proxy' }, + scheduledTaskService: { + async listDueTasks() { + calls.push('list'); + return [task]; + }, + async lockTask(id) { + calls.push(`lock:${id}`); + return task; + }, + async markTaskRunning(input) { + calls.push(`running:${input.id}`); + return { ...input, status: 'running' }; + }, + async markTaskSucceeded(input, payload) { + calls.push(`success:${input.id}:${payload.deliveryText}`); + return { ...input, status: 'completed' }; + }, + async markTaskFailed() { + calls.push('failed'); + }, + }, + scheduleService: { + async createUserNotification(input) { + notifications.push(input.notificationType); + }, + }, + notificationDispatcher: { + async sendScheduleNotification({ userId, text }) { + wechat.push({ userId, text }); + }, + }, + executeTask: async () => ({ + sessionId: 'session-1', + requestId: 'req-1', + deliveryText: '页面：https://example.com/news.html', + }), + logger: { warn() {}, info() {} }, + runOnStart: false, + }); + + await worker.runOnce(); + worker.stop(); + + assert.deepEqual(calls, [ + 'list', + 'lock:task-1', + 'running:task-1', + 'success:task-1:页面：https://example.com/news.html', + ]); + assert.deepEqual(notifications, ['scheduled_task_result']); + assert.equal(wechat.length, 1); + assert.match(wechat[0].text, /定时任务完成：每日新闻页/); +}); + +test('scheduled task worker marks failure after execution error', async () => { + const calls = []; + const task = { + id: 'task-2', + userId: 'user-2', + title: '失败任务', + recurrence: 'daily', + notifyChannel: 'web', + attempts: 3, + }; + const worker = startScheduledTaskWorker({ + intervalMs: 60_000, + userAuth: { id: 'user-auth' }, + tkmindProxy: { id: 'proxy' }, + scheduledTaskService: { + async listDueTasks() { + return [task]; + }, + async lockTask() { + return task; + }, + async markTaskRunning(input) { + return input; + }, + async markTaskSucceeded() { + calls.push('success'); + }, + async markTaskFailed(input, err, { maxAttempts }) { + calls.push(`failed:${maxAttempts}:${err.message}`); + return { ...input, status: 'failed', lastError: err.message }; + }, + }, + scheduleService: { + async createUserNotification(input) { + calls.push(`notify:${input.notificationType}`); + }, + }, + executeTask: async () => { + throw new Error('agent timeout'); + }, + maxAttempts: 3, + logger: { warn() {} }, + runOnStart: false, + }); + + await worker.runOnce(); + worker.stop(); + + assert.deepEqual(calls, ['failed:3:agent timeout', 'notify:scheduled_task_failed']); +}); diff --git a/scripts/verify-scheduled-task-automation.mjs b/scripts/verify-scheduled-task-automation.mjs new file mode 100644 index 0000000..37dbcb7 --- /dev/null +++ b/scripts/verify-scheduled-task-automation.mjs @@ -0,0 +1,277 @@ +#!/usr/bin/env node +/** + * 定时自动任务（Scheduled Task Automation）本地验证 + * + * 默认：DB + mock worker 全链路（无需 Portal 进程） + * 可选： + * VERIFY_SCHEDULED_TASK_LIVE=1 等待已开启 worker 的 Portal 到点执行（需 H5_SCHEDULED_TASK_WORKER_ENABLED=1） + * --due-seconds=90 创建 N 秒后执行的一次性任务（live 模式用） + */ +import assert from 'node:assert/strict'; +import crypto from 'node:crypto'; +import fs from 'node:fs'; +import mysql from 'mysql2/promise'; +import { migrateSchema } from '../db.mjs'; +import { parseScheduledTaskIntent, shouldUseScheduledTaskAutomation } from '../scheduled-task-intent.mjs'; +import { createScheduledTaskService } from '../scheduled-task-service.mjs'; +import { startScheduledTaskWorker } from '../scheduled-task-worker.mjs'; +import { getLocalParts, normalizeTimezone } from '../schedule-time.mjs'; + +const tz = 'Asia/Shanghai'; +const TEST_TITLE_PREFIX = 'VERIFY_SCHEDULED_TASK_'; +let passed = 0; +let failed = 0; + +function pass(label, detail = '') { + passed += 1; + console.log(`✔ ${label}${detail ? `: ${detail}` : ''}`); +} + +function fail(label, detail = '') { + failed += 1; + console.error(`✘ ${label}${detail ? `: ${detail}` : ''}`); +} + +function loadDatabaseUrl() { + if (process.env.DATABASE_URL) return process.env.DATABASE_URL; + const envPath = new URL('../.env', import.meta.url); + if (!fs.existsSync(envPath)) return null; + const envText = fs.readFileSync(envPath, 'utf8'); + const dbLine = envText.match(/^DATABASE_URL=(.+)$/m)?.[1]?.trim().replace(/^["']|["']$/g, ''); + return dbLine || null; +} + +function parseArgs(argv) { + const dueSecondsArg = argv.find((arg) => arg.startsWith('--due-seconds=')); + return { + dueSeconds: dueSecondsArg ? Number(dueSecondsArg.split('=')[1]) : 120, + }; +} + +function formatRunAtLocal(epochMs, timezone = tz) { + const parts = getLocalParts(epochMs, timezone); + return `${parts.year}-${String(parts.month).padStart(2, '0')}-${String(parts.day).padStart(2, '0')} ${String(parts.hour).padStart(2, '0')}:${String(parts.minute).padStart(2, '0')}`; +} + +function testIntentLayer() { + assert.equal(shouldUseScheduledTaskAutomation('每天6点帮我做今日新闻页面'), true); + assert.equal(parseScheduledTaskIntent('每天6点帮我做今日新闻页面').action, 'create_scheduled_task'); + assert.equal(shouldUseScheduledTaskAutomation('每天早上7点把当天待办发给我'), false); + pass('意图识别', '自动化 vs 待办 digest 边界正确'); +} + +async function cleanupTestRows(pool, userId) { + await pool.query( + `DELETE FROM h5_scheduled_tasks + WHERE user_id = ? AND title LIKE ?`, + [userId, `${TEST_TITLE_PREFIX}%`], + ); +} + +async function testDatabaseAndMockWorker(pool, userId, { dueSeconds }) { + await migrateSchema(pool); + pass('数据库 schema', 'h5_scheduled_tasks 已就绪'); + + const scheduledTaskService = createScheduledTaskService(pool, { + defaultTimezone: normalizeTimezone(tz), + }); + await cleanupTestRows(pool, userId); + + const runAtMs = Date.now() + Math.max(30, dueSeconds) * 1000; + const runAtLocal = formatRunAtLocal(runAtMs, tz); + const created = await scheduledTaskService.createTask({ + userId, + title: `${TEST_TITLE_PREFIX}mock_pipeline`, + taskSpec: '生成一段今日新闻摘要（验证脚本，无需写页面）', + recurrence: 'once', + runAtLocal, + notifyChannel: 'web', + sourceChannel: 'api', + sourceText: 'verify-scheduled-task-automation', + }); + if (created.recurrence === 'once' && created.nextRunAt > Date.now()) { + pass('创建一次性任务', `next_run_at=${new Date(created.nextRunAt).toLocaleString('zh-CN', { timeZone: tz })}`); + } else { + fail('创建一次性任务', `status=${created.status}, nextRunAt=${created.nextRunAt}`); + return; + } + + const notDue = await scheduledTaskService.listDueTasks({ now: Date.now() - 1000 }); + if (!notDue.some((row) => row.id === created.id)) { + pass('到期扫描（未到点）', '未来任务不会进入 due 列表'); + } else { + fail('到期扫描（未到点）', '未来任务不应被扫描为 due'); + } + + await pool.query( + `UPDATE h5_scheduled_tasks SET next_run_at = ?, status = 'active', locked_until = NULL WHERE id = ?`, + [Date.now() - 1000, created.id], + ); + + const notifications = []; + const worker = startScheduledTaskWorker({ + scheduledTaskService, + scheduleService: { + async createUserNotification(input) { + notifications.push(input); + }, + }, + userAuth: { id: 'mock-user-auth' }, + tkmindProxy: { id: 'mock-proxy' }, + executeTask: async (task) => ({ + sessionId: 'verify-session', + requestId: crypto.randomUUID(), + deliveryText: `验证交付：${task.title} 已在 ${new Date().toLocaleString('zh-CN', { timeZone: tz })} 完成。`, + }), + logger: { warn() {}, info() {} }, + runOnStart: false, + }); + + await worker.runOnce(); + worker.stop(); + + const [rows] = await pool.query( + `SELECT status, last_result_json, last_error FROM h5_scheduled_tasks WHERE id = ? LIMIT 1`, + [created.id], + ); + const row = rows[0]; + if (row?.status === 'completed') { + pass('mock worker 执行', '一次性任务 status=completed'); + } else { + fail('mock worker 执行', `status=${row?.status ?? 'missing'}, error=${row?.last_error ?? ''}`); + } + + if (notifications.some((item) => item.notificationType === 'scheduled_task_result')) { + pass('结果推送（mock）', 'scheduled_task_result 通知已写入'); + } else { + fail('结果推送（mock）', `notifications=${notifications.length}`); + } + + let lastResult = row?.last_result_json; + if (typeof lastResult === 'string') { + try { lastResult = JSON.parse(lastResult); } catch { lastResult = null; } + } + if (lastResult?.deliveryText) { + pass('执行结果落盘', String(lastResult.deliveryText).slice(0, 80)); + } else { + fail('执行结果落盘', 'last_result_json 缺少 deliveryText'); + } + + await cleanupTestRows(pool, userId); +} + +async function testLivePortalWorker(pool, userId, { dueSeconds }) { + const enabled = ['1', 'true', 'yes', 'on'].includes( + String(process.env.VERIFY_SCHEDULED_TASK_LIVE ?? '').toLowerCase(), + ); + if (!enabled) { + pass('Live Portal worker（跳过）', '设置 VERIFY_SCHEDULED_TASK_LIVE=1 可启用'); + return; + } + + const baseUrl = process.env.H5_PORT + ? `http://127.0.0.1:${process.env.H5_PORT}` + : 'http://127.0.0.1:8081'; + let portalOk = false; + try { + const res = await fetch(`${baseUrl}/api/status`, { signal: AbortSignal.timeout(3000) }); + portalOk = res.ok; + } catch { + portalOk = false; + } + if (!portalOk) { + pass('Live Portal worker（跳过）', `Portal 未运行：${baseUrl}/api/status 不可达`); + return; + } + pass('Live Portal 探活', baseUrl); + + const scheduledTaskService = createScheduledTaskService(pool, { + defaultTimezone: normalizeTimezone(tz), + }); + await cleanupTestRows(pool, userId); + + const waitSeconds = Math.max(30, Math.min(180, dueSeconds)); + const runAtMs = Date.now() + waitSeconds * 1000; + const created = await scheduledTaskService.createTask({ + userId, + title: `${TEST_TITLE_PREFIX}live_portal`, + taskSpec: '用三句话总结今天的热点新闻（验证脚本，回复文本即可，不要生成页面）', + recurrence: 'once', + runAtLocal: formatRunAtLocal(runAtMs, tz), + notifyChannel: 'web', + sourceChannel: 'api', + }); + pass('Live 任务已创建', `${waitSeconds}s 后执行，id=${created.id}`); + + console.log(`\n等待 Portal worker 执行（最多 ${waitSeconds + 120}s）…`); + console.log('请确认 Portal 进程已设置 H5_SCHEDULED_TASK_WORKER_ENABLED=1\n'); + + const deadline = Date.now() + (waitSeconds + 120) * 1000; + let terminal = null; + while (Date.now() < deadline) { + const [rows] = await pool.query( + `SELECT status, last_error, last_result_json, updated_at FROM h5_scheduled_tasks WHERE id = ? LIMIT 1`, + [created.id], + ); + terminal = rows[0]; + if (terminal?.status === 'completed' || terminal?.status === 'failed') break; + await new Promise((resolve) => setTimeout(resolve, 5000)); + } + + if (terminal?.status === 'completed') { + pass('Live Portal worker', `任务 completed，updated_at=${terminal.updated_at}`); + } else if (terminal?.status === 'failed') { + fail('Live Portal worker', terminal.last_error ?? 'failed'); + } else { + fail('Live Portal worker', `超时，最后 status=${terminal?.status ?? 'unknown'}`); + } + + await cleanupTestRows(pool, userId); +} + +async function main() { + const { dueSeconds } = parseArgs(process.argv.slice(2)); + console.log('=== 定时自动任务本地验证 ===\n'); + + testIntentLayer(); + + const dbUrl = loadDatabaseUrl(); + if (!dbUrl) { + fail('数据库场景', '未配置 DATABASE_URL'); + } else { + const pool = mysql.createPool(dbUrl); + try { + const username = process.env.VERIFY_SCHEDULED_TASK_USER ?? 'john8'; + const [users] = await pool.query( + 'SELECT id, username FROM h5_users WHERE username = ? LIMIT 1', + [username], + ); + if (!users[0]) { + const [fallback] = await pool.query('SELECT id, username FROM h5_users ORDER BY created_at ASC LIMIT 1'); + if (!fallback[0]) { + fail('数据库场景', '无可用测试用户'); + } else { + pass('测试用户', `${fallback[0].username} (${fallback[0].id})`); + await testDatabaseAndMockWorker(pool, fallback[0].id, { dueSeconds }); + await testLivePortalWorker(pool, fallback[0].id, { dueSeconds }); + } + } else { + pass('测试用户', `${users[0].username} (${users[0].id})`); + await testDatabaseAndMockWorker(pool, users[0].id, { dueSeconds }); + await testLivePortalWorker(pool, users[0].id, { dueSeconds }); + } + } finally { + await pool.end(); + } + } + + console.log('\n=== 汇总 ==='); + console.log(`通过: ${passed}`); + console.log(`失败: ${failed}`); + process.exit(failed > 0 ? 1 : 0); +} + +main().catch((err) => { + console.error(err); + process.exit(1); +}); diff --git a/server.mjs b/server.mjs index 792a658..ef1bbad 100644 --- a/server.mjs +++ b/server.mjs @@ -312,8 +312,10 @@ let wechatOAuthService = null; let wechatMpService = null; let notificationDispatcher = null; let scheduleService = null; +let scheduledTaskService = null; let feedbackService = null; let scheduleReminderWorker = null; +let scheduledTaskWorker = null; let llmProviderService = null; let wordFilterService = null; let authPool = null; @@ -358,6 +360,7 @@ async function bootstrapUserAuth() { mindSpaceRybbitConfig, ); scheduleService = domainServices.scheduleService; + scheduledTaskService = domainServices.scheduledTaskService; pageDataService = domainServices.pageDataService; pageDataPublicService = domainServices.pageDataPublicService; @@ -521,6 +524,8 @@ async function bootstrapUserAuth() { sessionAccess, tkmindProxy, scheduleService, + scheduledTaskService, + sessionSnapshotService, wechatScheduleLlmConfigService, llmProviderService, chatIntentRouter, @@ -541,6 +546,8 @@ async function bootstrapUserAuth() { integrationServices.notificationDispatcher; scheduleReminderWorker = integrationServices.scheduleReminderWorker; + scheduledTaskWorker = + integrationServices.scheduledTaskWorker; mindSpacePageEditSession = integrationServices.mindSpacePageEditSession; return true; diff --git a/server/portal-domain-services-bootstrap.mjs b/server/portal-domain-services-bootstrap.mjs index 84367e2..2e2873c 100644 --- a/server/portal-domain-services-bootstrap.mjs +++ b/server/portal-domain-services-bootstrap.mjs @@ -26,6 +26,7 @@ import { writebackPublications, } from '../plaza-tasks.mjs'; import { createScheduleService } from '../schedule-service.mjs'; +import { createScheduledTaskService } from '../scheduled-task-service.mjs'; import { createFeedbackService } from '../user-feedback.mjs'; import { PUBLISH_KEY_UUID } from '../user-publish.mjs'; import { initSchema } from '../db.mjs'; @@ -47,6 +48,7 @@ export async function bootstrapPortalDomainServices({ ensureMindSpaceConfigFn = ensureMindSpaceConfig, loadMindSpaceConfigFn = loadMindSpaceConfig, createScheduleServiceFn = createScheduleService, + createScheduledTaskServiceFn = createScheduledTaskService, createPageDataServiceFn = createPageDataService, createPageDataPublicServiceFn = createPageDataPublicService, @@ -114,6 +116,10 @@ export async function bootstrapPortalDomainServices({ defaultTimezone: env.H5_DEFAULT_TIMEZONE || 'Asia/Shanghai', }); + const scheduledTaskService = createScheduledTaskServiceFn(pool, { + defaultTimezone: + env.H5_DEFAULT_TIMEZONE || 'Asia/Shanghai', + }); const pageDataService = createPageDataServiceFn({ getUserAuth, getPool: () => pool, @@ -278,6 +284,7 @@ export async function bootstrapPortalDomainServices({ mindSpaceAnalyticsConfig: resolvedAnalyticsConfig, mindSpaceRybbitConfig: resolvedRybbitConfig, scheduleService, + scheduledTaskService, pageDataService, pageDataPublicService, feedbackService, diff --git a/server/portal-integration-services-bootstrap.mjs b/server/portal-integration-services-bootstrap.mjs index fc7d88d..b38466e 100644 --- a/server/portal-integration-services-bootstrap.mjs +++ b/server/portal-integration-services-bootstrap.mjs @@ -10,6 +10,7 @@ import { import { createPageEditSessionService } from '../mindspace-page-edit-session.mjs'; import { createNotificationDispatcher } from '../notification-dispatcher.mjs'; import { startScheduleReminderWorker } from '../schedule-reminder-worker.mjs'; +import { startScheduledTaskWorker } from '../scheduled-task-worker.mjs'; import { isPassiveCanaryRuntime } from './portal-runtime-role.mjs'; import { loadWechatMpModule } from '../wechat-mp-loader.mjs'; import { createToolGateway } from '../tool-gateway.mjs'; @@ -29,11 +30,12 @@ export async function bootstrapPortalIntegrationServices({ sessionAccess, tkmindProxy, scheduleService, + scheduledTaskService = null, + sessionSnapshotService = null, wechatScheduleLlmConfigService, llmProviderService, chatIntentRouter, systemDisclosurePolicyService, - sessionSnapshotService, mindSpaceAnalyticsConfig, mindSpaceRybbitConfig, subscriptionService, @@ -56,6 +58,8 @@ export async function bootstrapPortalIntegrationServices({ createNotificationDispatcher, startScheduleReminderWorkerFn = startScheduleReminderWorker, + startScheduledTaskWorkerFn = + startScheduledTaskWorker, createPageEditSessionServiceFn = createPageEditSessionService, createToolGatewayFn = createToolGateway, @@ -266,6 +270,27 @@ export async function bootstrapPortalIntegrationServices({ logger.log('Schedule reminder worker enabled'); } + let scheduledTaskWorker = null; + if ( + !isPassiveCanaryRuntime(env) && + env.H5_SCHEDULED_TASK_WORKER_ENABLED === '1' && + scheduledTaskService && + userAuth && + tkmindProxy + ) { + scheduledTaskWorker = + startScheduledTaskWorkerFn({ + scheduledTaskService, + scheduleService, + userAuth, + tkmindProxy, + sessionSnapshotService, + notificationDispatcher, + logger, + }); + logger.log('Scheduled task worker enabled'); + } + let subscriptionExpiryTimer = null; if (subscriptionService && !isPassiveCanaryRuntime(env)) { subscriptionExpiryTimer = setIntervalFn( @@ -323,6 +348,7 @@ export async function bootstrapPortalIntegrationServices({ wechatMpService, notificationDispatcher, scheduleReminderWorker, + scheduledTaskWorker, subscriptionExpiryTimer, mindSpacePageEditSession, }; diff --git a/skills-registry.mjs b/skills-registry.mjs index dfc73e7..e6df4aa 100644 --- a/skills-registry.mjs +++ b/skills-registry.mjs @@ -21,6 +21,7 @@ export const DEFAULT_USER_SKILLS = { 'search-enhanced': true, [EXCEL_ANALYST_SKILL_NAME]: false, 'schedule-assistant': true, + 'scheduled-task-automation': true, 'service-integration-smoke': true, 'form-builder': true, 'table-viewer': true, diff --git a/skills-registry.test.mjs b/skills-registry.test.mjs index b480275..88012a6 100644 --- a/skills-registry.test.mjs +++ b/skills-registry.test.mjs @@ -18,6 +18,7 @@ test('lists static-page-publish in platform catalog', () => { assert.ok(catalog.some((item) => item.name === 'static-page-publish')); assert.ok(catalog.some((item) => item.name === 'page-data-collect')); assert.ok(catalog.some((item) => item.name === 'schedule-assistant')); + assert.ok(catalog.some((item) => item.name === 'scheduled-task-automation')); assert.ok(catalog.some((item) => item.name === 'service-integration-smoke')); assert.ok(catalog.some((item) => item.name === 'product-campaign-page')); assert.ok(catalog.some((item) => item.name === 'long-image-download')); @@ -68,6 +69,7 @@ test('DEFAULT_USER_SKILLS enables common platform skills', () => { assert.equal(DEFAULT_USER_SKILLS.web, true); assert.equal(DEFAULT_USER_SKILLS.search, true); assert.equal(DEFAULT_USER_SKILLS['schedule-assistant'], true); + assert.equal(DEFAULT_USER_SKILLS['scheduled-task-automation'], true); assert.equal(DEFAULT_USER_SKILLS['service-integration-smoke'], true); assert.equal(DEFAULT_USER_SKILLS['form-builder'], true); assert.equal(DEFAULT_USER_SKILLS['table-viewer'], true); diff --git a/skills/scheduled-task-automation/SKILL.md b/skills/scheduled-task-automation/SKILL.md new file mode 100644 index 0000000..d11086b --- /dev/null +++ b/skills/scheduled-task-automation/SKILL.md @@ -0,0 +1,67 @@ +--- +name: scheduled-task-automation +description: 处理定时自动任务（Scheduled Automation）：澄清执行时间与任务内容，写入订阅；到点由系统自动执行并推送结果。不处理纯待办/提醒。 +--- + +# 定时自动任务（Scheduled Task Automation） + +这个 skill 处理**到点自动执行并交付结果**的定时任务，与 `schedule-assistant`（待办/提醒）严格分离。 + +## 适用范围 + +- 创建一次性或循环的定时自动任务（Scheduled Task / Recurring Automation） +- 查询、取消当前用户的定时任务 +- 澄清缺失的执行时间或任务内容 + +## 边界 + +1. **不**处理纯提醒、待办记录、日程查询 —— 那些交给 `schedule-assistant`。 +2. **不**在当前对话里立即执行长任务；本 skill 只负责**写入定时订阅**。 +3. 只能操作当前用户的数据；工具成功返回前，不能说「已经设置好了」。 +4. 信息不完整时先追问，禁止臆造时间或任务内容。 + +## 可用工具 + +- `scheduled_task_create` +- `scheduled_task_list` +- `scheduled_task_cancel` + +## 必填信息（创建前必须齐全） + +| 字段 | 说明 | +|------|------| +| `taskSpec` | 到时具体执行什么（给后续 Agent 的说明） | +| `recurrence` | `once` 一次性 / `daily` 每天 / `weekly` 每周 | +| 执行时间 | `once` 用 `runAtLocal`；`daily`/`weekly` 用 `hour`+`minute`（weekly 还需 `weekday`） | + +## 时间写入（必守） + +1. **禁止**自行估算 Unix 毫秒时间戳。 +2. 一次性任务用 `runAtLocal`（`YYYY-MM-DD HH:mm`，用户时区）。 +3. 循环任务用 `hour`（0-23）+ `minute`（0-59）；每周任务额外传 `weekday`（0=周日 … 6=周六）。 +4. 写入后可用 `scheduled_task_list` 核对。 + +## 工作规则 + +1. 先判断用户是要**创建**、**查询**还是**取消**定时自动任务。 +2. 缺时间或缺任务内容时，一次只追问最小必要信息。 +3. 创建成功后，用用户能理解的话说明： + - 何时执行 + - 执行什么 + - 是一次性还是循环 + - 结果会通过哪个通道推送（默认微信 + 站内） +4. 取消时优先用 `taskId`；用户只说任务名时可传 `titleMatch`。 +5. 若提示里给出了 `sourceMessageId`，调用 `scheduled_task_create` 时必须原样传入。 + +## 回复要求 + +- 简短、直接 +- 成功时说明任务摘要、时间与 recurrence +- 失败时明确没写入成功，并给出下一步 + +## 示例 + +用户：每天 6 点帮我做今日新闻页面 + +助手：（信息齐全后调用 `scheduled_task_create`） +已设置循环任务：每天 06:00（北京时间）自动搜索并生成「今日新闻页」，完成后推送链接。 diff --git a/user-memory-profile.mjs b/user-memory-profile.mjs index eadaa0a..e4324fa 100644 --- a/user-memory-profile.mjs +++ b/user-memory-profile.mjs @@ -265,6 +265,17 @@ export function buildSessionMemoryEntries({ ].join('\n'), }); } + if ((scheduleTools?.available_tools ?? []).includes('scheduled_task_create')) { + entries.push({ + title: 'TKMind 定时自动任务规则', + content: [ + '创建 Scheduled Automation 时：', + '- 使用 scheduled_task_create；一次性任务传 runAtLocal，循环任务传 hour/minute/weekday。', + '- 禁止自行估算 Unix 毫秒；本 skill 只写入订阅，不在当前对话立即执行长任务。', + '- 写入后调用 scheduled_task_list 核对。', + ].join('\n'), + }); + } if (!hasMemoryStore(sessionPolicy)) { return entries; diff --git a/wechat/prompts/chat-general.mjs b/wechat/prompts/chat-general.mjs index 09046f1..e518e86 100644 --- a/wechat/prompts/chat-general.mjs +++ b/wechat/prompts/chat-general.mjs @@ -4,6 +4,7 @@ import { isPageDataIntent, } from '../../chat-skills.mjs'; import { shouldUseScheduleAssistant } from '../../schedule-intent.mjs'; +import { shouldUseScheduledTaskAutomation } from '../../scheduled-task-intent.mjs'; import { buildCurrentTimeAgentPrefix } from '../../user-memory-profile.mjs'; import { isPageGenerateText, wantsDocxDownload } from '../intent/patterns.mjs'; import { buildWechatStandaloneImageInstruction } from '../image-generation-policy.mjs'; @@ -81,6 +82,17 @@ export function buildWechatAgentPrompt(intent, { '', ].filter(Boolean).join('\n') : ''; + const scheduledTaskAutomationHint = shouldUseScheduledTaskAutomation(agentText) + ? [ + '【定时自动任务要求】这条消息是 Scheduled Automation（到点自动执行并推送结果），不是待办提醒。', + '开始前先加载 `scheduled-task-automation` skill，并严格按 skill 边界执行。', + '先澄清缺失的执行时间或 taskSpec，再调用 scheduled_task_create / scheduled_task_list / scheduled_task_cancel。', + '禁止在当前对话立即执行长任务；工具成功返回后才能确认已设置。', + '一次性任务用 runAtLocal；循环任务用 hour/minute/weekday，禁止自行估算 Unix 毫秒。', + intent?.msgId ? `调用 scheduled_task_create 时必须传入 sourceMessageId: ${intent.msgId}` : '', + '', + ].filter(Boolean).join('\n') + : ''; const imageGenerationHint = buildWechatStandaloneImageInstruction(imagePolicy, { sourceMessageId: intent?.msgId, }); @@ -92,6 +104,7 @@ export function buildWechatAgentPrompt(intent, { pageDataRepairHint, currentTimeHint, scheduleAssistantHint, + scheduledTaskAutomationHint, imageGenerationHint, '【微信服务号语音消息】用户通过语音输入，以下是微信识别结果。', '', @@ -105,6 +118,7 @@ export function buildWechatAgentPrompt(intent, { docxDownloadHint, currentTimeHint, scheduleAssistantHint, + scheduledTaskAutomationHint, '【微信服务号图片消息】用户发送了图片。', '如用户没有明确要求，请先根据图片内容给出简短理解，并询问下一步。', '', @@ -132,6 +146,7 @@ export function buildWechatAgentPrompt(intent, { return [ currentTimeHint, scheduleAssistantHint, + scheduledTaskAutomationHint, '【微信服务号位置消息】用户发送了当前位置。', location.label ? `地址：${location.label}` : '', location.latitude !== null && location.latitude !== undefined @@ -150,6 +165,7 @@ export function buildWechatAgentPrompt(intent, { return [ currentTimeHint, scheduleAssistantHint, + scheduledTaskAutomationHint, '【微信服务号链接消息】用户分享了链接。', link.title ? `标题：${link.title}` : '', link.description ? `描述：${link.description}` : '', @@ -171,6 +187,7 @@ export function buildWechatAgentPrompt(intent, { if (sessionPageContinuationHint) lines.push(sessionPageContinuationHint); if (pagePublishHint) lines.push(pagePublishHint); if (scheduleAssistantHint) lines.push(scheduleAssistantHint); + if (scheduledTaskAutomationHint) lines.push(scheduledTaskAutomationHint); if (imageGenerationHint) lines.push(imageGenerationHint); lines.push( '【微信服务号新消息】请只回答下面这条用户消息，不要主动延续无关的历史话题。',