diff --git a/agent/tools/scheduler/scheduler_service.py b/agent/tools/scheduler/scheduler_service.py index 1f4bc6fb..ee36dd11 100644 --- a/agent/tools/scheduler/scheduler_service.py +++ b/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: """ diff --git a/channel/web/static/js/console.js b/channel/web/static/js/console.js index 94899f2a..8acec37d 100644 --- a/channel/web/static/js/console.js +++ b/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 = `${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 = `${t('task_run_started')}`; + setTimeout(() => { + button.innerHTML = originalHtml; + button.disabled = false; + }, 1500); + }).catch(() => { + button.innerHTML = `${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() {
{t('task_channel_locked')}
+ {runStatus &&{runStatus}
} {error &&{error}
}
diff --git a/docs/zh/tools/scheduler.mdx b/docs/zh/tools/scheduler.mdx
index 2648116f..bf4f6dc7 100644
--- a/docs/zh/tools/scheduler.mdx
+++ b/docs/zh/tools/scheduler.mdx
@@ -35,6 +35,12 @@ description: 创建和管理定时任务
- "明天下午 3 点提醒我开会"
- "查看所有定时任务"
+## 测试执行已有任务
+
+在经过身份认证的 Web 控制台或 Desktop 应用中打开任务,点击 **立即执行**。确认发送后,CowAgent 会立即按任务已配置的通道和接收者执行一次。这个入口只允许用户手动触发,不会暴露给 Agent 的 scheduler tool。
+
+禁用状态的任务也可手动执行,且不会因此被启用、删除或重新调度,原有 `next_run_at` 保持不变。同一任务正在按计划执行时,不允许重复手动执行。
+
diff --git a/tests/test_scheduler_run_now.py b/tests/test_scheduler_run_now.py
new file mode 100644
index 00000000..b8301bc2
--- /dev/null
+++ b/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
diff --git a/tests/test_scheduler_web_update.py b/tests/test_scheduler_web_update.py
index 2b423ce2..8a7c240d 100644
--- a/tests/test_scheduler_web_update.py
+++ b/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",