From f7a0184ea2447863e4c93ca7e4d5f77a4292d33b Mon Sep 17 00:00:00 2001 From: AaronZ345 <34849476+AaronZ345@users.noreply.github.com> Date: Fri, 17 Jul 2026 11:58:44 +0800 Subject: [PATCH] feat(scheduler): run existing tasks immediately --- agent/tools/scheduler/scheduler_service.py | 63 ++++++++++++++- agent/tools/scheduler/scheduler_tool.py | 30 +++++++- docs/tools/scheduler.mdx | 8 ++ docs/zh/tools/scheduler.mdx | 5 ++ tests/test_scheduler_run_now.py | 89 ++++++++++++++++++++++ 5 files changed, 190 insertions(+), 5 deletions(-) create mode 100644 tests/test_scheduler_run_now.py 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/agent/tools/scheduler/scheduler_tool.py b/agent/tools/scheduler/scheduler_tool.py index de8802a1..bdf271b1 100644 --- a/agent/tools/scheduler/scheduler_tool.py +++ b/agent/tools/scheduler/scheduler_tool.py @@ -24,6 +24,7 @@ class SchedulerTool(BaseTool): "⚠️ 重要：仅当需要「定时/提醒/每天/每周/X分钟后/X点」等延迟或周期执行时才使用此工具。" "使用方法：\n" "- 创建：action='create', name='任务名', message/ai_task='内容', schedule_type='once/interval/cron', schedule_value='...'\n" + "- 立即执行：action='run', task_id='任务ID'（不改变原调度时间）\n" "- 查询：action='list' / action='get', task_id='任务ID'\n" "- 管理：action='delete/enable/disable', task_id='任务ID'\n\n" "调度类型：\n" @@ -37,12 +38,12 @@ class SchedulerTool(BaseTool): "properties": { "action": { "type": "string", - "enum": ["create", "list", "get", "delete", "enable", "disable"], - "description": "操作类型: create(创建), list(列表), get(查询), delete(删除), enable(启用), disable(禁用)" + "enum": ["create", "run", "list", "get", "delete", "enable", "disable"], + "description": "操作类型: create(创建), run(立即执行), list(列表), get(查询), delete(删除), enable(启用), disable(禁用)" }, "task_id": { "type": "string", - "description": "任务ID (用于 get/delete/enable/disable 操作)" + "description": "任务ID (用于 run/get/delete/enable/disable 操作)" }, "name": { "type": "string", @@ -88,7 +89,7 @@ class SchedulerTool(BaseTool): Args: params: Dictionary containing: - - action: Operation type (create/list/get/delete/enable/disable) + - action: Operation type (create/run/list/get/delete/enable/disable) - Other parameters depending on action Returns: @@ -105,6 +106,9 @@ class SchedulerTool(BaseTool): if action == "create": result = self._create_task(**kwargs) return ToolResult.success(result) + elif action == "run": + result = self._run_task(**kwargs) + return ToolResult.success(result) elif action == "list": result = self._list_tasks(**kwargs) return ToolResult.success(result) @@ -303,6 +307,24 @@ class SchedulerTool(BaseTool): self.task_store.delete_task(task_id) return f"✅ 任务 '{task['name']}' ({task_id}) 已删除" + + def _run_task(self, **kwargs) -> str: + """Queue an existing task for immediate execution.""" + task_id = kwargs.get("task_id") + if not task_id: + return "错误: 缺少任务ID (task_id)" + + task = self.task_store.get_task(task_id) + if not task: + return f"错误: 任务 '{task_id}' 不存在" + + from agent.tools.scheduler.integration import get_scheduler_service + service = get_scheduler_service() + if not service: + return "错误: 定时任务系统未运行" + + service.run_task_now(task_id) + return f"✅ 任务 '{task['name']}' ({task_id}) 已开始立即执行，原调度时间保持不变" def _enable_task(self, **kwargs) -> str: """Enable a task""" diff --git a/docs/tools/scheduler.mdx b/docs/tools/scheduler.mdx index 18c211bf..47135c67 100644 --- a/docs/tools/scheduler.mdx +++ b/docs/tools/scheduler.mdx @@ -34,6 +34,14 @@ Create and manage scheduled tasks with natural language: - "Check server status every 2 hours" - "Remind me about the meeting tomorrow at 3 PM" - "Show all scheduled tasks" +- "Run task a1b2c3d4 now" + +## Run an existing task now + +Use `action="run"` with a task ID to queue an immediate test run. 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. diff --git a/docs/zh/tools/scheduler.mdx b/docs/zh/tools/scheduler.mdx index 2648116f..44cd8f5e 100644 --- a/docs/zh/tools/scheduler.mdx +++ b/docs/zh/tools/scheduler.mdx @@ -34,6 +34,11 @@ description: 创建和管理定时任务 - "每隔 2 小时检查一下服务器状态" - "明天下午 3 点提醒我开会" - "查看所有定时任务" +- "立即执行任务 a1b2c3d4" + +## 立即执行已有任务 + +使用任务 ID 调用 `action="run"`，可以排队进行一次即时测试执行。禁用状态的任务也可手动执行，且不会因此被启用、删除或重新调度，原有 `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..71cf990b --- /dev/null +++ b/tests/test_scheduler_run_now.py @@ -0,0 +1,89 @@ +"""Tests for immediate execution of persisted scheduler tasks.""" + +import threading +import time +from datetime import datetime, timedelta +from unittest.mock import Mock, patch + +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_queues_existing_task(): + store = Mock() + store.get_task.return_value = {"id": "task-1", "name": "refresh index"} + service = Mock() + tool = SchedulerTool() + tool.task_store = store + + with patch( + "agent.tools.scheduler.integration.get_scheduler_service", + return_value=service, + ): + result = tool.execute({"action": "run", "task_id": "task-1"}) + + assert result.status == "success" + service.run_task_now.assert_called_once_with("task-1")