Merge pull request #2958 from AaronZ345/feat/scheduler-run-now

feat(scheduler): add user-confirmed run-now controls

agent/tools/scheduler/scheduler_service.py

@@ -41,6 +41,8 @@ class SchedulerService:
         self.running = False
         self.thread = None
         self._lock = threading.Lock()
+        self._execution_lock = threading.Lock()
+        self._active_task_ids = set()
     
     def start(self):
         """Start the scheduler service"""
@@ -85,7 +87,15 @@ class SchedulerService:
             try:
                 if self._is_task_due(task, now):
                     logger.info(f"[Scheduler] Executing task: {task['id']} - {task['name']}")
-                    ok = self._execute_task(task)
+                    if not self._claim_task(task['id']):
+                        logger.info(
+                            f"[Scheduler] Task {task['id']} is already running; skipping this tick"
+                        )
+                        continue
+                    try:
+                        ok = self._execute_task(task)
+                    finally:
+                        self._release_task(task['id'])
                     if not ok:
                         # Leave next_run_at as-is so the next loop retries.
                         # Cron tasks within the catch-up window will keep
@@ -106,6 +116,57 @@ class SchedulerService:
                         logger.info(f"[Scheduler] One-time task completed and removed: {task['id']}")
             except Exception as e:
                 logger.error(f"[Scheduler] Error processing task {task.get('id')}: {e}")
+
+    def run_task_now(self, task_id: str) -> None:
+        """Queue one immediate execution without changing the task schedule.
+
+        Disabled and one-time tasks may be run manually for testing. The
+        stored ``next_run_at`` remains unchanged, so a manual run never
+        consumes or delays the next scheduled occurrence.
+
+        Raises:
+            ValueError: if the task does not exist.
+            RuntimeError: if the same task is already executing.
+        """
+        task = self.task_store.get_task(task_id)
+        if not task:
+            raise ValueError(f"Task '{task_id}' not found")
+        if not self._claim_task(task_id):
+            raise RuntimeError(f"Task '{task_id}' is already running")
+
+        def _run():
+            now = datetime.now()
+            try:
+                logger.info(f"[Scheduler] Manually executing task: {task_id} - {task.get('name', '')}")
+                ok = self._execute_task(task)
+                if ok:
+                    self.task_store.update_task(task_id, {
+                        "last_run_at": now.isoformat(),
+                        "last_manual_run_at": now.isoformat(),
+                    })
+                    logger.info(f"[Scheduler] Manual execution completed: {task_id}")
+                else:
+                    logger.warning(f"[Scheduler] Manual execution failed: {task_id}")
+            finally:
+                self._release_task(task_id)
+
+        threading.Thread(
+            target=_run,
+            daemon=True,
+            name=f"scheduler-manual-{task_id}",
+        ).start()
+
+    def _claim_task(self, task_id: str) -> bool:
+        """Prevent scheduled and manual runs of the same task from overlapping."""
+        with self._execution_lock:
+            if task_id in self._active_task_ids:
+                return False
+            self._active_task_ids.add(task_id)
+            return True
+
+    def _release_task(self, task_id: str) -> None:
+        with self._execution_lock:
+            self._active_task_ids.discard(task_id)
     
     def _is_task_due(self, task: dict, now: datetime) -> bool:
         """

channel/web/static/js/console.js

@@ -217,6 +217,11 @@ const I18N = {
         task_delete_btn: '删除任务',
         task_delete_confirm_title: '删除定时任务',
         task_delete_confirm_msg: '确定删除该定时任务吗？此操作无法撤销。',
+        task_run_now: '立即执行',
+        task_run_confirm_title: '立即执行任务',
+        task_run_confirm_msg: '该任务会立即向已配置的通道和接收者发送内容。是否继续？',
+        task_run_started: '已开始执行',
+        task_run_failed: '执行失败',
         logs_title: '日志', logs_desc: '实时日志输出 (run.log)',
         logs_live: '实时', logs_coming_msg: '日志流即将在此提供。将连接 run.log 实现类似 tail -f 的实时输出。',
         new_chat: '新对话',
@@ -464,6 +469,11 @@ const I18N = {
         task_delete_btn: '刪除任務',
         task_delete_confirm_title: '刪除定時任務',
         task_delete_confirm_msg: '確定刪除該定時任務嗎？此操作無法撤銷。',
+        task_run_now: '立即執行',
+        task_run_confirm_title: '立即執行任務',
+        task_run_confirm_msg: '該任務會立即向已設定的通道和接收者傳送內容。是否繼續？',
+        task_run_started: '已開始執行',
+        task_run_failed: '執行失敗',
         logs_title: '日誌', logs_desc: '實時日誌輸出 (run.log)',
         logs_live: '實時', logs_coming_msg: '日誌流即將在此提供。將連線 run.log 實現類似 tail -f 的實時輸出。',
         new_chat: '新對話',
@@ -710,6 +720,11 @@ const I18N = {
         task_delete_btn: 'Delete Task',
         task_delete_confirm_title: 'Delete Task',
         task_delete_confirm_msg: 'Delete this scheduled task? This action cannot be undone.',
+        task_run_now: 'Run now',
+        task_run_confirm_title: 'Run task now',
+        task_run_confirm_msg: 'This task will immediately send to its configured channel and receiver. Continue?',
+        task_run_started: 'Run started',
+        task_run_failed: 'Run failed',
         logs_title: 'Logs', logs_desc: 'Real-time log output (run.log)',
         logs_live: 'Live', logs_coming_msg: 'Log streaming will be available here. Connects to run.log for real-time output similar to tail -f.',
         new_chat: 'New Chat',
@@ -7935,6 +7950,38 @@ function refreshTasksView() {
         btn.disabled = false;
     }, 500);
 }
+
+function runTaskNow(task, button) {
+    showConfirmDialog({
+        title: t('task_run_confirm_title'),
+        message: `${task.name || task.id}: ${t('task_run_confirm_msg')}`,
+        okText: t('task_run_now'),
+        onConfirm: () => {
+            const originalHtml = button.innerHTML;
+            button.disabled = true;
+            button.innerHTML = `<i class="fas fa-spinner fa-spin mr-1"></i>${t('task_run_now')}`;
+            fetch('/api/scheduler/run', {
+                method: 'POST',
+                headers: {'Content-Type': 'application/json'},
+                body: JSON.stringify({task_id: task.id})
+            }).then(r => r.json()).then(res => {
+                if (res.status !== 'success') throw new Error(res.message || t('task_run_failed'));
+                button.innerHTML = `<i class="fas fa-check mr-1"></i>${t('task_run_started')}`;
+                setTimeout(() => {
+                    button.innerHTML = originalHtml;
+                    button.disabled = false;
+                }, 1500);
+            }).catch(() => {
+                button.innerHTML = `<i class="fas fa-triangle-exclamation mr-1"></i>${t('task_run_failed')}`;
+                setTimeout(() => {
+                    button.innerHTML = originalHtml;
+                    button.disabled = false;
+                }, 2000);
+            });
+        }
+    });
+}
+
 function loadTasksView() {
     if (tasksLoaded) return;
     fetch('/api/scheduler').then(r => r.json()).then(data => {
@@ -7996,11 +8043,19 @@ function loadTasksView() {
                 <div class="flex items-center gap-4 text-xs text-slate-400 dark:text-slate-500">
                     <span><i class="fas fa-clock mr-1"></i>${currentLang === 'zh' ? '下次执行' : 'Next run'}: ${nextRun}</span>
                     <div class="flex-1"></div>
+                    <button type="button" class="task-run-now px-2 py-1 rounded-md text-primary-500 hover:bg-primary-50 dark:hover:bg-primary-500/10 transition-colors">
+                        <i class="fas fa-play mr-1"></i>${t('task_run_now')}
+                    </button>
                     <label class="relative inline-flex items-center cursor-pointer" for="${toggleId}">
                         <input type="checkbox" id="${toggleId}" class="sr-only peer" ${isEnabled ? 'checked' : ''}>
                         <div class="w-9 h-5 bg-slate-200 peer-focus:outline-none rounded-full peer peer-checked:after:translate-x-full peer-checked:after:border-white after:content-[''] after:absolute after:top-[2px] after:left-[2px] after:bg-white after:rounded-full after:h-4 after:w-4 after:transition-all peer-checked:bg-primary-500 dark:bg-slate-600 dark:peer-checked:bg-primary-500"></div>
                     </label>
                 </div>`;
+            const runButton = card.querySelector('.task-run-now');
+            runButton.addEventListener('click', function(e) {
+                e.stopPropagation();
+                runTaskNow(task, runButton);
+            });
             const checkbox = card.querySelector('#' + toggleId);
             checkbox.addEventListener('change', function() {
                 const newEnabled = this.checked;

channel/web/web_channel.py

@@ -1309,6 +1309,7 @@ class WebChannel(ChatChannel):
             '/api/knowledge/action', 'KnowledgeActionHandler',
             '/api/knowledge/import', 'KnowledgeImportHandler',
             '/api/scheduler', 'SchedulerHandler',
+            '/api/scheduler/run', 'SchedulerRunHandler',
             '/api/scheduler/toggle', 'SchedulerToggleHandler',
             '/api/scheduler/update', 'SchedulerUpdateHandler',
             '/api/scheduler/delete', 'SchedulerDeleteHandler',
@@ -4537,6 +4538,34 @@ class SchedulerHandler:
             return json.dumps({"status": "error", "message": str(e)})
 
 
+class SchedulerRunHandler:
+    def POST(self):
+        _require_auth()
+        web.header('Content-Type', 'application/json; charset=utf-8')
+        try:
+            body = json.loads(web.data())
+            task_id = body.get("task_id")
+            if not task_id:
+                return json.dumps({"status": "error", "message": "task_id required"})
+
+            from agent.tools.scheduler.integration import get_scheduler_service
+            service = get_scheduler_service()
+            if service is None:
+                return json.dumps({
+                    "status": "error",
+                    "message": "Scheduler service is not running",
+                })
+
+            service.run_task_now(task_id)
+            return json.dumps({
+                "status": "success",
+                "message": f"Task '{task_id}' queued for immediate execution",
+            }, ensure_ascii=False)
+        except Exception as e:
+            logger.error(f"[WebChannel] Scheduler manual run error: {e}")
+            return json.dumps({"status": "error", "message": str(e)})
+
+
 class SchedulerToggleHandler:
     def POST(self):
         _require_auth()

desktop/src/renderer/src/api/client.ts

@@ -372,6 +372,13 @@ class ApiClient {
     return data.tasks
   }
 
+  async runTask(taskId: string): Promise<ApiResult> {
+    return this.request('/api/scheduler/run', {
+      method: 'POST',
+      body: JSON.stringify({ task_id: taskId }),
+    })
+  }
+
   async toggleTask(taskId: string, enabled: boolean): Promise<{ status: string; task: SchedulerTask }> {
     return this.request('/api/scheduler/toggle', {
       method: 'POST',

desktop/src/renderer/src/i18n.ts

@@ -352,6 +352,10 @@ const translations: Record<string, Record<string, string>> = {
     task_delete: '删除',
     task_delete_confirm: '确定删除该任务吗？此操作不可撤销。',
     task_save_error: '保存失败',
+    task_run_now: '立即执行',
+    task_run_confirm: '该任务会立即向已配置的通道和接收者发送内容。是否继续？',
+    task_run_started: '任务已开始执行',
+    task_run_error: '执行失败',
     logs_title: '日志',
     logs_desc: '实时日志输出 (run.log)',
     logs_live: '实时',
@@ -740,6 +744,10 @@ const translations: Record<string, Record<string, string>> = {
     task_delete: 'Delete',
     task_delete_confirm: 'Delete this task? This cannot be undone.',
     task_save_error: 'Failed to save',
+    task_run_now: 'Run now',
+    task_run_confirm: 'This task will immediately send to its configured channel and receiver. Continue?',
+    task_run_started: 'Task run started',
+    task_run_error: 'Failed to run task',
     logs_title: 'Logs',
     logs_desc: 'Real-time log output (run.log)',
     logs_live: 'Live',

desktop/src/renderer/src/pages/TasksPage.tsx

@@ -1,5 +1,5 @@
 import React, { useEffect, useState } from 'react'
-import { Loader2, Clock, CalendarClock } from 'lucide-react'
+import { Loader2, Clock, CalendarClock, Play } from 'lucide-react'
 import { useNavigate } from 'react-router-dom'
 import { t } from '../i18n'
 import apiClient from '../api/client'
@@ -165,6 +165,8 @@ const TaskEditModal: React.FC<{
   const [actionType, setActionType] = useState<TaskAction['type']>(task.action.type || 'send_message')
   const [content, setContent] = useState(task.action.content || task.action.task_description || '')
   const [saving, setSaving] = useState(false)
+  const [running, setRunning] = useState(false)
+  const [runStatus, setRunStatus] = useState('')
   const [error, setError] = useState('')
 
   const buildSchedule = (): TaskSchedule => {
@@ -209,6 +211,22 @@ const TaskEditModal: React.FC<{
     }
   }
 
+  const runNow = async () => {
+    if (!window.confirm(t('task_run_confirm'))) return
+    setRunning(true)
+    setRunStatus('')
+    setError('')
+    try {
+      const result = await apiClient.runTask(task.id)
+      if (result.status !== 'success') throw new Error(result.message || t('task_run_error'))
+      setRunStatus(t('task_run_started'))
+    } catch (e) {
+      setError(e instanceof Error ? e.message : t('task_run_error'))
+    } finally {
+      setRunning(false)
+    }
+  }
+
   return (
     <Modal
       open
@@ -219,6 +237,10 @@ const TaskEditModal: React.FC<{
           <Btn variant="danger" onClick={del} disabled={saving} className="mr-auto">
             {t('task_delete')}
           </Btn>
+          <Btn variant="ghost" onClick={runNow} disabled={saving || running}>
+            {running ? <Loader2 size={14} className="inline animate-spin mr-1" /> : <Play size={14} className="inline mr-1" />}
+            {t('task_run_now')}
+          </Btn>
           <Btn variant="ghost" onClick={onClose} disabled={saving}>
             {t('task_cancel')}
           </Btn>
@@ -298,6 +320,7 @@ const TaskEditModal: React.FC<{
       )}
       <p className="text-xs text-content-tertiary">{t('task_channel_locked')}</p>
 
+      {runStatus && <p className="text-xs text-success">{runStatus}</p>}
       {error && <p className="text-xs text-danger">{error}</p>}
     </Modal>
   )

docs/tools/scheduler.mdx

@@ -35,6 +35,17 @@ Create and manage scheduled tasks with natural language:
 - "Remind me about the meeting tomorrow at 3 PM"
 - "Show all scheduled tasks"
 
+## Test-run an existing task
+
+Open the task in the authenticated Web console or Desktop app and click **Run
+now**. After you confirm the delivery, CowAgent queues one immediate execution
+to the task's configured channel and receiver. This user-initiated action is not
+exposed to the agent's scheduler tool.
+
+Manual execution works for disabled tasks and does not enable, delete, or
+reschedule the task. Its original `next_run_at` remains unchanged. A task cannot
+be run manually while the same task is already executing on schedule.
+
 <Frame>
   <img src="https://cdn.link-ai.tech/doc/20260202195402.png" width="800" />
 </Frame>

docs/zh/tools/scheduler.mdx

@@ -35,6 +35,12 @@ description: 创建和管理定时任务
 - "明天下午 3 点提醒我开会"
 - "查看所有定时任务"
 
+## 测试执行已有任务
+
+在经过身份认证的 Web 控制台或 Desktop 应用中打开任务，点击 **立即执行**。确认发送后，CowAgent 会立即按任务已配置的通道和接收者执行一次。这个入口只允许用户手动触发，不会暴露给 Agent 的 scheduler tool。
+
+禁用状态的任务也可手动执行，且不会因此被启用、删除或重新调度，原有 `next_run_at` 保持不变。同一任务正在按计划执行时，不允许重复手动执行。
+
 <Frame>
   <img src="https://cdn.link-ai.tech/doc/20260202195402.png" width="800" />
 </Frame>

tests/test_scheduler_run_now.py

@@ -0,0 +1,77 @@
+"""Tests for immediate execution of persisted scheduler tasks."""
+
+import threading
+import time
+from datetime import datetime, timedelta
+
+import pytest
+
+from agent.tools.scheduler.scheduler_service import SchedulerService
+from agent.tools.scheduler.scheduler_tool import SchedulerTool
+from agent.tools.scheduler.task_store import TaskStore
+
+
+def _task(task_id="task-1", enabled=False):
+    return {
+        "id": task_id,
+        "name": "refresh index",
+        "enabled": enabled,
+        "created_at": datetime.now().isoformat(),
+        "updated_at": datetime.now().isoformat(),
+        "next_run_at": (datetime.now() + timedelta(hours=2)).isoformat(),
+        "schedule": {"type": "interval", "seconds": 7200},
+        "action": {"type": "send_message", "content": "done"},
+    }
+
+
+def test_manual_run_executes_disabled_task_without_moving_schedule(tmp_path):
+    store = TaskStore(str(tmp_path / "tasks.json"))
+    task = _task()
+    store.add_task(task)
+    completed = threading.Event()
+
+    def execute(received):
+        assert received["id"] == "task-1"
+        completed.set()
+        return True
+
+    service = SchedulerService(store, execute)
+    service.run_task_now("task-1")
+
+    assert completed.wait(timeout=2)
+    deadline = time.time() + 2
+    while time.time() < deadline:
+        updated = store.get_task("task-1")
+        if updated.get("last_manual_run_at"):
+            break
+        time.sleep(0.01)
+
+    updated = store.get_task("task-1")
+    assert updated["next_run_at"] == task["next_run_at"]
+    assert updated["enabled"] is False
+    assert updated["last_run_at"] == updated["last_manual_run_at"]
+
+
+def test_manual_run_rejects_duplicate_execution(tmp_path):
+    store = TaskStore(str(tmp_path / "tasks.json"))
+    store.add_task(_task())
+    release = threading.Event()
+    started = threading.Event()
+
+    def execute(_task_data):
+        started.set()
+        release.wait(timeout=2)
+        return True
+
+    service = SchedulerService(store, execute)
+    service.run_task_now("task-1")
+    assert started.wait(timeout=2)
+    with pytest.raises(RuntimeError, match="already running"):
+        service.run_task_now("task-1")
+    release.set()
+
+
+def test_scheduler_tool_does_not_expose_manual_execution():
+    actions = SchedulerTool.params["properties"]["action"]["enum"]
+
+    assert "run" not in actions

tests/test_scheduler_web_update.py

@@ -4,7 +4,8 @@ import json
 import sys
 import types
 from datetime import datetime, timedelta
-from unittest.mock import patch
+from pathlib import Path
+from unittest.mock import Mock, patch
 
 from agent.tools.scheduler.task_store import TaskStore
 
@@ -28,7 +29,9 @@ if "web" not in sys.modules:
     )
     sys.modules["web"] = web_stub
 
-from channel.web.web_channel import SchedulerUpdateHandler
+from channel.web import web_channel
+
+SchedulerUpdateHandler = web_channel.SchedulerUpdateHandler
 
 
 def _store_task(tmp_path, action):
@@ -54,6 +57,57 @@ def _post_update(tmp_path, payload):
         return json.loads(SchedulerUpdateHandler().POST())
 
 
+def test_web_manual_run_is_authenticated_and_delegates_to_scheduler():
+    assert hasattr(web_channel, "SchedulerRunHandler")
+
+    service = Mock()
+    with patch("channel.web.web_channel._require_auth") as require_auth, \
+         patch("channel.web.web_channel.web.header"), \
+         patch("channel.web.web_channel.web.data", return_value=b'{"task_id":"task-1"}'), \
+         patch("agent.tools.scheduler.integration.get_scheduler_service", return_value=service):
+        response = json.loads(web_channel.SchedulerRunHandler().POST())
+
+    require_auth.assert_called_once_with()
+    service.run_task_now.assert_called_once_with("task-1")
+    assert response == {
+        "status": "success",
+        "message": "Task 'task-1' queued for immediate execution",
+    }
+
+
+def test_web_manual_run_rejects_unavailable_scheduler():
+    assert hasattr(web_channel, "SchedulerRunHandler")
+
+    with patch("channel.web.web_channel._require_auth"), \
+         patch("channel.web.web_channel.web.header"), \
+         patch("channel.web.web_channel.web.data", return_value=b'{"task_id":"task-1"}'), \
+         patch("agent.tools.scheduler.integration.get_scheduler_service", return_value=None):
+        response = json.loads(web_channel.SchedulerRunHandler().POST())
+
+    assert response == {
+        "status": "error",
+        "message": "Scheduler service is not running",
+    }
+
+
+def test_manual_run_is_exposed_by_explicit_web_and_desktop_controls():
+    root = Path(__file__).parents[1]
+    web_source = (root / "channel/web/web_channel.py").read_text(encoding="utf-8")
+    web_console = (root / "channel/web/static/js/console.js").read_text(encoding="utf-8")
+    desktop_client = (root / "desktop/src/renderer/src/api/client.ts").read_text(encoding="utf-8")
+    desktop_page = (root / "desktop/src/renderer/src/pages/TasksPage.tsx").read_text(encoding="utf-8")
+
+    assert "'/api/scheduler/run', 'SchedulerRunHandler'" in web_source
+    assert "function runTaskNow(task, button)" in web_console
+    assert "fetch('/api/scheduler/run'" in web_console
+    web_run = web_console[web_console.index("function runTaskNow(task, button)"):]
+    assert "showConfirmDialog({" in web_run[:2500]
+    assert "async runTask(taskId: string)" in desktop_client
+    assert "'/api/scheduler/run'" in desktop_client
+    assert "const runNow = async ()" in desktop_page
+    assert "window.confirm(t('task_run_confirm'))" in desktop_page
+
+
 def test_web_edit_preserves_hidden_agent_action_fields(tmp_path):
     store = _store_task(tmp_path, {
         "type": "agent_task",