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.
 
 <Frame>
   <img src="https://cdn.link-ai.tech/doc/20260202195402.png" width="800" />
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` 保持不变。同一任务正在按计划执行时，不允许重复手动执行。
 
 <Frame>
   <img src="https://cdn.link-ai.tech/doc/20260202195402.png" width="800" />
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")