diff --git a/CLAUDE.md b/CLAUDE.md
index 6642d73..87ba397 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -169,6 +169,8 @@ Before drawing any conclusion from a tool call, verify:
 - __Don't conflate "test contamination" with "tool bug"__ — if a tool worked once and then fails, check the test state before assuming the tool is broken. Window switching, manual user interaction between calls, and stale controllers are common contamination sources.
 - __`save_captured_image` writes into project bundle, not MaaMCP data dir__ — destination is `<bundle_root>/image/<subcategory>/<name>.png` (the path TemplateMatch's `template` field reads from). `bundle_root` is the directory passed to `Resource.post_bundle()`. For MAAGC it's `assets/resource/base/`; for MaaFramework sample it's `<repo>/sample/resource`. Default `overwrite=False` protects existing templates — pass `True` explicitly when updating.
 - __`benchmark_node` measures wall-clock, not per-node timing__ — returned `latency_ms` is full `post_task → TaskDetail` time, including entry recognition overhead (~50-200ms). For per-node estimate subtract that baseline. `mean_score=None` with `successes=0` means the node never hit — threshold/ROI/template mismatch.
+- __`run_pipeline` / `benchmark_node` accept per-run `pipeline_override`__ — field-level node overrides passed to `post_task` (same mechanism as interface.json's `pipeline_override`): tighten `roi`, tweak `expected`/`threshold`/`timeout` for single-node verification without editing pipeline files. Applies to one run only — the Resource is untouched. Node names not present in the loaded files show up in `warnings` (typo'd names silently no-op, so check them). `benchmark_node` rejects overriding `next` on entry/target (would break its isolation chain).
+- __`run_pipeline` supports tool-side `timeout_seconds`__ — polls the task and calls `post_stop()` on expiry, returning `status="timeout"` + partial node details instead of blocking the MCP call indefinitely (a node that never matches burns its own `timeout`, 20s by default). Note: MaaFramework marks a stopped task's own status as succeeded — the tool reports `"timeout"` explicitly instead of trusting it. Recommended 5-15s for single-node verification; `None` (default) keeps the old unbounded behavior.
 
 ### Pipeline node tuning loop
 
@@ -180,7 +182,7 @@ When a TemplateMatch / OCR / ColorMatch node isn't reliable, iterate this loop (
 4. `save_captured_image(cropped_path, bundle_root, subcategory, name)` → promote it to a TemplateMatch template
 5. In pipeline JSON: `"recognition": "TemplateMatch", "template": "<subcategory>/<name>.png"`
 6. `benchmark_node(cid, pipeline_path, node=<name>, iterations=10..50)` → inspect `mean_score`, `latency_ms`, `all_results_samples`
-7. If `mean_score` < 0.85 or `successes < iterations`: tighten ROI (smaller `region`), raise `threshold`, or refresh the template with a fresh capture
+7. If `mean_score` < 0.85 or `successes < iterations`: tighten ROI (smaller `region`), raise `threshold`, or refresh the template with a fresh capture — try candidate values via `pipeline_override={"<name>": {"roi": [...], "threshold": ...}}` first (no file edits), then write the winning values back into the pipeline JSON
 8. Repeat 2-7 until stable
 
 For MaaMCP-side pipeline infra testing, see `tests/test_dbg_pipeline.py` (gated by `@pytest.mark.integration`; skips if `MaaDbgControlUnit` DLL isn't shipped).
diff --git a/CLAUDE_CN.md b/CLAUDE_CN.md
index 5142779..4dc4411 100644
--- a/CLAUDE_CN.md
+++ b/CLAUDE_CN.md
@@ -168,6 +168,8 @@ OCR 模型和截图存储在平台特定的目录中：
 - __别把"测试污染"误判成"工具 bug"__——工具用过一次后又失败，先检查测试状态（是否切窗、是否手动操作、controller 是否失效），再怀疑工具本身。
 - __`save_captured_image` 写入项目 bundle，不是 MaaMCP 数据目录__——目标是 `<bundle_root>/image/<子分类>/<元素名>.png`（TemplateMatch 的 `template` 字段读取路径）。`bundle_root` 是传给 `Resource.post_bundle()` 的目录：MAAGC 是 `assets/resource/base/`；MaaFramework sample 是 `<repo>/sample/resource`。默认 `overwrite=False` 保护已有模板，更新时显式传 `True`。
 - __`benchmark_node` 测的是 wall-clock，不是单节点耗时__——返回的 `latency_ms` 是 `post_task → TaskDetail` 总耗时，含 entry 识别开销（~50-200ms）。想估节点本身耗时减掉这段基线。`mean_score=None` 且 `successes=0` 表示一次都没命中——通常是阈值/ROI/模板漂移的信号。
+- __`run_pipeline` / `benchmark_node` 支持单次 `pipeline_override`__——字段级节点覆盖，直接传给 `post_task`（与 interface.json 的 `pipeline_override` 同机制）：单节点验证时收紧 `roi`、调 `expected`/`threshold`/`timeout`，不用改 pipeline 文件。只对单次运行生效——Resource 不被污染。覆盖的节点名若不在已加载文件中会出现在 `warnings`（拼错节点名时覆盖静默不生效，记得检查）。`benchmark_node` 拒绝覆盖 entry/target 的 `next`（会破坏隔离链路）。
+- __`run_pipeline` 支持工具侧 `timeout_seconds`__——轮询任务状态，超时调 `post_stop()`，返回 `status="timeout"` + 部分节点详情，不再无限阻塞 MCP 调用（识别不命中的节点会烧满自己的 `timeout`，默认 20s）。注意：MaaFramework 把被 stop 的任务自身 status 标记为 succeeded——工具显式报 `"timeout"`，不信任它。单节点验证建议 5-15s；`None`（默认）保持旧的不限时行为。
 
 ### Pipeline 节点调参循环
 
@@ -179,7 +181,7 @@ TemplateMatch / OCR / ColorMatch 节点不稳定时，按这个循环迭代（is
 4. `save_captured_image(cropped_path, bundle_root, subcategory, name)` → 提到 TemplateMatch 模板
 5. pipeline JSON 里写：`"recognition": "TemplateMatch", "template": "<subcategory>/<name>.png"`
 6. `benchmark_node(cid, pipeline_path, node=<name>, iterations=10..50)` → 看 `mean_score`、`latency_ms`、`all_results_samples`
-7. `mean_score < 0.85` 或 `successes < iterations`：收紧 ROI（缩 `region`）、抬高 `threshold`、或重新截一张更准的模板
+7. `mean_score < 0.85` 或 `successes < iterations`：收紧 ROI（缩 `region`）、抬高 `threshold`、或重新截一张更准的模板——先用 `pipeline_override={"<name>": {"roi": [...], "threshold": ...}}` 免改文件试参，确定后再把最终值写回 pipeline JSON
 8. 重复 2-7 直到稳定
 
 MaaMCP 侧 pipeline infra 集成测试见 `tests/test_dbg_pipeline.py`（标 `@pytest.mark.integration`；缺 `MaaDbgControlUnit` DLL 时自动 skip）。
diff --git a/README.md b/README.md
index cd38930..5f6fcc3 100644
--- a/README.md
+++ b/README.md
@@ -77,7 +77,7 @@ Talk is cheap, 请看: **[🎞️ Bilibili 视频演示](https://www.bilibili.co
 - `get_pipeline_protocol` - 获取 Pipeline 协议文档
 - `save_pipeline` - 保存 Pipeline JSON 到文件（支持新建和更新）
 - `load_pipeline` - 读取已有的 Pipeline 文件
-- `run_pipeline` - 运行 Pipeline 并返回执行结果（支持单/多文件、Custom action agent 自动启动）
+- `run_pipeline` - 运行 Pipeline 并返回执行结果（支持单/多文件、Custom action agent 自动启动、单次 `pipeline_override` 参数覆盖、工具侧 `timeout_seconds` 超时）
 
 ### 🛑 Pipeline 终止
 
@@ -380,6 +380,28 @@ run_pipeline(
 - 加载是原子的：所有预校验通过后才写入 Resource
 - 节点驻留：已加载的节点持续驻留，切换 pipeline 集时调用 `clear_pipeline_resources()` 重置
 
+### 单节点快速验证（pipeline_override + timeout_seconds）
+
+调试单个节点时，不必反复改 pipeline 文件，也不必忍受整屏识别慢扫或无限阻塞：
+
+```python
+run_pipeline(
+    controller_id,
+    "main.json",
+    entry="点击设置",                        # 只从被测节点进入
+    start_agent=False,                       # 只测识别，不拉起 CustomAction 链
+    pipeline_override={                      # 单次生效：不写文件、不污染 Resource
+        "点击设置": {"roi": [520, 20, 200, 80], "timeout": 3000}
+    },
+    timeout_seconds=10,                      # 工具侧超时：超过 10s 自动 post_stop
+)
+```
+
+- `pipeline_override` 与 interface.json 的同名机制一致：**按字段合并**到已加载节点，
+  也可覆盖 `expected` / `threshold` / `enabled` 等任意字段；ROI 越准单次识别越快
+- 超时返回 `status="timeout"` + 超时前已执行的节点详情，方便定位挂在哪个节点
+- `benchmark_node` 同样支持 `pipeline_override`，调参循环免改文件
+
 ## 注意事项
 
 📌 **Windows 自动化限制**：
diff --git a/README_EN.md b/README_EN.md
index d437838..6808fb2 100644
--- a/README_EN.md
+++ b/README_EN.md
@@ -77,7 +77,7 @@ Talk is cheap, see: **[🎞️ Bilibili Video Demo](https://www.bilibili.com/vid
 - `get_pipeline_protocol` - Get Pipeline protocol documentation
 - `save_pipeline` - Save Pipeline JSON to file (supports creating and updating)
 - `load_pipeline` - Load an existing Pipeline file
-- `run_pipeline` - Run Pipeline and return execution results (single/multi-file, auto-start Custom action agent)
+- `run_pipeline` - Run Pipeline and return execution results (single/multi-file, auto-start Custom action agent, per-run `pipeline_override`, tool-side `timeout_seconds`)
 
 ### 🛑 Pipeline Termination
 
@@ -282,6 +282,27 @@ After Pipeline generation, AI automatically validates and optimizes:
 
 If the Pipeline logic itself needs adjustment, AI can re-execute automation operations and combine old and new experiences to generate a more robust Pipeline.
 
+### Fast Single-Node Verification (pipeline_override + timeout_seconds)
+
+When debugging a single node, there is no need to repeatedly edit the pipeline file, suffer slow full-screen recognition, or let the call block forever:
+
+```python
+run_pipeline(
+    controller_id,
+    "main.json",
+    entry="ClickSettings",                  # enter from the node under test
+    start_agent=False,                      # recognition only, no CustomAction chain
+    pipeline_override={                     # per-run only: no file edits, Resource untouched
+        "ClickSettings": {"roi": [520, 20, 200, 80], "timeout": 3000}
+    },
+    timeout_seconds=10,                     # tool-side cap: auto post_stop after 10s
+)
+```
+
+- `pipeline_override` uses the same mechanism as interface.json's field: **field-level merge** into loaded nodes; any field (`expected` / `threshold` / `enabled` / ...) can be overridden — and the tighter the ROI, the faster each recognition
+- On timeout the result carries `status="timeout"` plus the nodes executed before the stop, showing exactly where it hung
+- `benchmark_node` accepts the same `pipeline_override`, making tuning loops file-edit-free
+
 ### Example Output
 
 ```json
diff --git a/maa_mcp/pipeline_tools.py b/maa_mcp/pipeline_tools.py
index a046b5f..7f5ccf2 100644
--- a/maa_mcp/pipeline_tools.py
+++ b/maa_mcp/pipeline_tools.py
@@ -8,6 +8,7 @@
 """
 
 import json
+import math
 import time
 from dataclasses import dataclass, field
 from datetime import datetime
@@ -608,6 +609,124 @@ def _validate_entry(
     return entry
 
 
+def _validate_timeout_seconds(timeout_seconds: Optional[float]) -> None:
+    """校验工具侧超时参数：None（不限时）或正的有限数值（秒）。
+
+    NaN / inf 必须拒绝：deadline = monotonic() + NaN 之后 "now >= deadline"
+    永远为 False，超时分支不可达——等于静默禁用超时还带 CPU 空转轮询。
+    JSON 层（jiter/pydantic）接受 NaN/Infinity/1e400，所以这里必须自己拦。
+
+    Raises:
+        ValueError: 非 None 且不是正的有限数值（bool、NaN、±inf、
+            超出 float 范围的巨大整数都视为非法）。
+    """
+    if timeout_seconds is None:
+        return
+    if isinstance(timeout_seconds, bool) or not isinstance(timeout_seconds, (int, float)):
+        raise ValueError(
+            f"timeout_seconds 必须是正数（秒）或 None，实际: {timeout_seconds!r}"
+        )
+    try:
+        value = float(timeout_seconds)
+    except OverflowError:
+        value = float("inf")
+    if not math.isfinite(value):
+        raise ValueError(
+            f"timeout_seconds 必须是有限数值（NaN/inf 会静默禁用超时），"
+            f"实际: {timeout_seconds!r}"
+        )
+    if value <= 0:
+        raise ValueError(f"timeout_seconds 必须 > 0，实际: {timeout_seconds!r}")
+
+
+def _validate_pipeline_override(pipeline_override: Optional[dict]) -> dict:
+    """校验 pipeline_override 结构：dict[节点名(str) → 字段 dict]。
+
+    只做结构校验，不校验节点名是否存在（未知节点名的提示见
+    _unknown_override_nodes——覆盖可以指向 bundle 资源里的节点或新增节点，
+    所以未知名是 warning 不是 error）。
+
+    Returns:
+        规范化后的 override dict；None 输入返回空 dict。
+
+    Raises:
+        ValueError: 结构非法（顶层非 dict / 键非字符串 / 节点值非 dict）。
+    """
+    if pipeline_override is None:
+        return {}
+    if not isinstance(pipeline_override, dict):
+        raise ValueError(
+            f"pipeline_override 必须是 dict（节点名 → 字段 dict），"
+            f"实际类型: {type(pipeline_override).__name__}"
+        )
+    for name, fields in pipeline_override.items():
+        if not isinstance(name, str) or not name:
+            raise ValueError(
+                f"pipeline_override 的键必须是非空字符串节点名，实际: {name!r}"
+            )
+        if not isinstance(fields, dict):
+            raise ValueError(
+                f"pipeline_override[{name!r}] 必须是字段 dict，"
+                f"实际类型: {type(fields).__name__}"
+            )
+    return pipeline_override
+
+
+def _unknown_override_nodes(pipeline_override: dict, merged: dict) -> List[str]:
+    """找出 override 中不在本次加载文件节点表里的节点名，生成提示 warning。
+
+    未知节点名不阻断执行：它可能指向 bundle 资源中已有的节点，也可能是
+    override 新增的节点（MaaFramework 均支持）。warning 只用于帮助发现拼写错误
+    （拼错节点名时覆盖会静默不生效，是最常见的踩坑点）。
+    """
+    unknown = sorted(n for n in pipeline_override if n not in merged)
+    return [
+        f"pipeline_override 节点 {n!r} 不在本次加载的文件节点表中"
+        f"（可能来自 bundle 资源或为新增节点；若是想覆盖已加载节点，请检查拼写）"
+        for n in unknown
+    ]
+
+
+_TIMEOUT_POLL_INTERVAL_S = 0.05
+
+
+def _wait_task_with_timeout(
+    tasker: Any,
+    task_job: Any,
+    timeout_seconds: Optional[float],
+    poll_interval: float = _TIMEOUT_POLL_INTERVAL_S,
+) -> tuple[Any, bool]:
+    """等待任务完成；超过 timeout_seconds 则 post_stop() 主动停止。
+
+    timeout_seconds=None 时保持旧行为：task_job.wait() 无限阻塞。
+    超时路径：tasker.post_stop().wait() 确保任务终止后，再取一次
+    TaskDetail——它携带超时前已执行的部分节点详情，便于诊断挂在哪个节点。
+
+    Args:
+        tasker: maafw Tasker（超时时用它 post_stop）。
+        task_job: tasker.post_task 返回的 TaskJob。
+        timeout_seconds: 工具侧超时（秒）；None 表示不限时。
+        poll_interval: 轮询间隔（秒）。
+
+    Returns:
+        (task_detail 或 None, timed_out)
+    """
+    if timeout_seconds is None:
+        return task_job.wait().get(), False
+
+    deadline = time.monotonic() + timeout_seconds
+    while True:
+        if task_job.done:
+            return task_job.get(), False
+        if time.monotonic() >= deadline:
+            break
+        time.sleep(poll_interval)
+
+    tasker.post_stop().wait()
+    # stop 完成后任务已终止，wait() 立即返回；detail 含部分执行的节点
+    return task_job.wait().get(), True
+
+
 def _parse_task_status(status: Any) -> str:
     """将 MaaFramework TaskDetail.status 映射为可读字符串。"""
     if status.succeeded:
@@ -623,16 +742,11 @@ def _parse_task_status(status: Any) -> str:
     return str(status)
 
 
-def _build_run_result(
-    file_dicts: List[tuple[str, dict]],
-    entry_node: str,
-    merged: dict,
-    task_detail: Any,
-    conflicts: List[str],
-    on_conflict: ConflictStrategy,
-) -> PipelineLoadResult:
-    """组装 run_pipeline 的返回结果。"""
+def _extract_nodes_info(task_detail: Any) -> List[dict]:
+    """从 TaskDetail 抽取节点执行详情列表；detail 为 None 或无节点时返回空 list。"""
     nodes_info: List[dict] = []
+    if task_detail is None:
+        return nodes_info
     if hasattr(task_detail, "nodes") and task_detail.nodes:
         for node in task_detail.nodes:
             node_info: dict = {}
@@ -643,10 +757,29 @@ def _build_run_result(
             if hasattr(node, "name"):
                 node_info["name"] = node.name
             nodes_info.append(node_info)
+    return nodes_info
 
-    warnings: List[str] = []
+
+def _conflict_warnings(
+    conflicts: List[str], on_conflict: ConflictStrategy
+) -> List[str]:
+    """OVERWRITE 模式下把冲突节点名转成 warning 文案；其余模式返回空 list。"""
     if on_conflict == ConflictStrategy.OVERWRITE and conflicts:
-        warnings = [f"节点冲突（后文件覆盖前文件）: {n}" for n in conflicts]
+        return [f"节点冲突（后文件覆盖前文件）: {n}" for n in conflicts]
+    return []
+
+
+def _build_run_result(
+    file_dicts: List[tuple[str, dict]],
+    entry_node: str,
+    merged: dict,
+    task_detail: Any,
+    conflicts: List[str],
+    on_conflict: ConflictStrategy,
+    extra_warnings: Optional[List[str]] = None,
+) -> PipelineLoadResult:
+    """组装 run_pipeline 的返回结果。"""
+    warnings = _conflict_warnings(conflicts, on_conflict) + list(extra_warnings or [])
 
     return PipelineLoadResult(
         success=bool(task_detail.status.succeeded),
@@ -655,7 +788,7 @@ def _build_run_result(
         entry=entry_node,
         status=_parse_task_status(task_detail.status),
         task_id=task_detail.task_id,
-        nodes=nodes_info,
+        nodes=_extract_nodes_info(task_detail),
         warnings=warnings,
     )
 
@@ -818,6 +951,22 @@ def save_pipeline(
         - "strict" (默认): 检测到任何节点冲突立即返回错误，不写入 Resource
         - "overwrite": 后加载的整节点覆盖先加载的（与 MaaFramework 行为一致），
                        冲突节点名会出现在返回的 warnings 中
+    - pipeline_override: 单次运行的节点参数覆盖（可选）。dict 格式：节点名 → 字段 dict。
+      按字段级合并到已加载的 pipeline（与 interface.json 的 pipeline_override 同机制），
+      只对本次执行生效——不写入 Resource、不修改 pipeline 文件。
+      典型用法——单节点快速验证（只截 ROI 区域验证，避免整屏 OCR 慢扫）：
+        run_pipeline(cid, "p.json", entry="MyNode", start_agent=False,
+                     pipeline_override={"MyNode": {"roi": [x, y, w, h], "timeout": 3000}},
+                     timeout_seconds=10)
+      ROI 越准单次识别越快；配合节点级 timeout（毫秒）与工具级 timeout_seconds 双保险。
+      也可覆盖 expected / threshold / enabled 等任意节点字段，或新增临时节点。
+      覆盖的节点名若不在本次加载的文件节点表中，会出现在返回的 warnings 里
+      （可能来自 bundle 资源或为新增节点；主要帮助发现拼写错误——拼错时覆盖会静默不生效）。
+    - timeout_seconds: 工具侧执行超时（秒，可选；默认 None = 不限时，保持旧行为）。
+      超过该时长自动调用 post_stop() 停止任务，并返回 status="timeout" 的结构化结果
+      （含超时前已执行的部分节点详情），不再让 MCP 调用无限阻塞。
+      单节点验证建议 5~15；完整业务流程按需放宽或不设。
+      超时只停当前任务；后台 agent 子进程如需一并清理，另行调用 stop_pipeline()。
 
     返回值：
     - 成功/失败统一返回 PipelineLoadResult 序列化后的 dict，包含以下字段：
@@ -825,7 +974,7 @@ def save_pipeline(
       - files: 实际加载的文件绝对路径列表
       - node_count: 合并后写入 Resource 的节点总数
       - entry: 入口节点名称
-      - status: 执行状态字符串（"succeeded" | "failed" | "running" | "pending" | "done"）
+      - status: 执行状态字符串（"succeeded" | "failed" | "running" | "pending" | "done" | "timeout"）
       - task_id: 任务 ID
       - nodes: 节点详情列表
         - name: 节点名称
@@ -881,7 +1030,16 @@ def run_pipeline(
     resource_path: Optional[str] = None,
     on_conflict: str = ConflictStrategy.STRICT.value,
     start_agent: bool = True,
+    pipeline_override: Optional[dict] = None,
+    timeout_seconds: Optional[float] = None,
 ) -> dict | str:
+    # 0. 新参数结构预校验（纯校验，先于任何全局状态修改）
+    try:
+        _validate_timeout_seconds(timeout_seconds)
+        override_dict = _validate_pipeline_override(pipeline_override)
+    except ValueError as e:
+        return f"参数错误: {e}"
+
     # 如果传入了 resource_path，添加它以便 get_or_create_resource 加载该路径
     if resource_path:
         add_resource_path(resource_path)
@@ -953,9 +1111,31 @@ def run_pipeline(
                 f"identifier={agent_ctx.identifier}"
             )
 
-    # 6. 执行任务
-    task_job = tasker.post_task(entry_node)
-    task_detail = task_job.wait().get()
+    # 6. 执行任务（可选：单次 pipeline_override + 工具侧超时）
+    override_warnings = _unknown_override_nodes(override_dict, merged)
+    task_job = tasker.post_task(entry_node, override_dict)
+    task_detail, timed_out = _wait_task_with_timeout(
+        tasker, task_job, timeout_seconds
+    )
+
+    if timed_out:
+        return PipelineLoadResult(
+            success=False,
+            files=[f for f, _ in file_dicts],
+            node_count=len(merged),
+            entry=entry_node,
+            status="timeout",
+            task_id=getattr(task_detail, "task_id", 0) if task_detail else 0,
+            nodes=_extract_nodes_info(task_detail),
+            warnings=_conflict_warnings(conflicts, strategy) + override_warnings,
+            error=(
+                f"任务执行超过 timeout_seconds={timeout_seconds}s，"
+                "已调用 post_stop() 主动停止。常见原因：某节点识别一直未命中"
+                "（roi 偏了 / expected 不匹配 / threshold 过高）。"
+                "建议先 screencap 看实际画面，再用 pipeline_override 收紧 roi 重试。"
+                "nodes 字段包含超时前已执行的部分节点详情。"
+            ),
+        ).to_dict()
 
     if not task_detail:
         return "任务执行失败，无法获取执行详情"
@@ -968,6 +1148,7 @@ def run_pipeline(
         task_detail=task_detail,
         conflicts=conflicts,
         on_conflict=strategy,
+        extra_warnings=override_warnings,
     ).to_dict()
 
 
@@ -1078,6 +1259,7 @@ class BenchmarkRunResult:
     - latency_ms: 每次 post_task 到拿到 TaskDetail 的总耗时（毫秒，整数）
     - all_results_samples: 前 3 次的 all_results 采样（每项 box + score），
       方便快速排查识别漂移
+    - warnings: 警告信息（如 pipeline_override 中的未知节点名）；为空时不序列化
     """
 
     node: str
@@ -1088,9 +1270,10 @@ class BenchmarkRunResult:
     mean_score: Optional[float]
     latency_ms: List[int]
     all_results_samples: List[List[dict]]
+    warnings: List[str] = field(default_factory=list)
 
     def to_dict(self) -> dict:
-        return {
+        result = {
             "node": self.node,
             "iterations": self.iterations,
             "successes": self.successes,
@@ -1100,6 +1283,9 @@ def to_dict(self) -> dict:
             "latency_ms": list(self.latency_ms),
             "all_results_samples": [list(s) for s in self.all_results_samples],
         }
+        if self.warnings:
+            result["warnings"] = list(self.warnings)
+        return result
 
 
 _BENCHMARK_DONE_NODE = "_BenchmarkDone"
@@ -1193,12 +1379,14 @@ def _summarize_iteration(node_detail: Any, target_name: str) -> tuple[Optional[f
 def _aggregate_benchmark(
     target_name: str,
     per_iter: List[tuple[Optional[float], List[dict], int]],
+    warnings: Optional[List[str]] = None,
 ) -> BenchmarkRunResult:
     """聚合多次迭代结果。
 
     Args:
         target_name: 被测节点名
         per_iter: [(score, sample, latency_ms), ...]
+        warnings: 附加警告信息（可选）
 
     Returns:
         BenchmarkRunResult 实例。
@@ -1215,6 +1403,7 @@ def _aggregate_benchmark(
         mean_score=(sum(scores) / len(scores)) if scores else None,
         latency_ms=latencies,
         all_results_samples=samples,
+        warnings=list(warnings or []),
     )
 
 
@@ -1225,17 +1414,19 @@ def _benchmark_node_impl(
     entry: Optional[str] = None,
     iterations: int = 10,
     resource_path: Optional[str] = None,
+    pipeline_override: Optional[dict] = None,
 ) -> dict | str:
     """benchmark_node 的核心实现（不含 @mcp.tool 装饰，便于单测 + 复用）。
 
     行为：
-    1. 校验 iterations ∈ [1, 1000]
+    1. 校验 iterations ∈ [1, 1000] + pipeline_override 结构
     2. 加载 + 合并 pipeline（用 OVERWRITE 策略，benchmark_node 视角下冲突无所谓）
     3. 解析 entry（默认用首个文件的第一个 key）
     4. 校验 node / entry 都在 merged 中
     5. 加载 Resource（若有 resource_path）+ Tasker
     6. 构造 override pipeline（强制 entry → target → done）
-    7. 跑 iterations 次，每次记录 wall-clock latency + target.score + 采样
+    7. 跑 iterations 次（每次 post_task 附带 pipeline_override 做字段级覆盖），
+       记录 wall-clock latency + target.score + 采样
     8. 聚合返回 BenchmarkRunResult
     """
     if not isinstance(iterations, int) or iterations < 1 or iterations > 1000:
@@ -1243,6 +1434,11 @@ def _benchmark_node_impl(
             f"参数错误: iterations 必须是 1..1000 的整数，实际: {iterations!r}"
         )
 
+    try:
+        override_dict = _validate_pipeline_override(pipeline_override)
+    except ValueError as e:
+        return f"参数错误: {e}"
+
     # 1) 规范化 + 预校验 + 合并
     try:
         files = _normalize_paths(pipeline_path)
@@ -1265,6 +1461,17 @@ def _benchmark_node_impl(
     if entry_node not in merged:
         return f"入口节点 {entry_node!r} 不存在（可用节点: {sorted(merged.keys())}）"
 
+    # pipeline_override 不允许改 entry / node 的 next：
+    # benchmark 靠强制 entry → node → done 链路隔离下游副作用，
+    # 覆盖 next 会破坏隔离（post_task 级覆盖优先于 Resource 级 override）
+    for protected in (entry_node, node):
+        if protected in override_dict and "next" in override_dict[protected]:
+            return (
+                f"参数错误: pipeline_override 不允许覆盖 {protected!r} 的 next 字段"
+                "（会破坏 benchmark 的 entry → node → done 隔离链路）"
+            )
+    override_warnings = _unknown_override_nodes(override_dict, merged)
+
     # 2) Resource + Tasker
     if resource_path:
         add_resource_path(resource_path)
@@ -1283,11 +1490,11 @@ def _benchmark_node_impl(
     if not resource.override_pipeline(override):
         return "override_pipeline 写入失败"
 
-    # 4) 跑 iterations
+    # 4) 跑 iterations（post_task 级 pipeline_override 按字段合并，不动 Resource）
     per_iter: List[tuple[Optional[float], List[dict], int]] = []
     for _ in range(iterations):
         start = time.perf_counter()
-        detail = tasker.post_task(entry_node).wait().get()
+        detail = tasker.post_task(entry_node, override_dict).wait().get()
         elapsed_ms = int(round((time.perf_counter() - start) * 1000))
         target_detail = None
         if detail and getattr(detail, "nodes", None):
@@ -1298,7 +1505,7 @@ def _benchmark_node_impl(
         score, sample = _summarize_iteration(target_detail, node)
         per_iter.append((score, sample, elapsed_ms))
 
-    return _aggregate_benchmark(node, per_iter).to_dict()
+    return _aggregate_benchmark(node, per_iter, warnings=override_warnings).to_dict()
 
 
 @mcp.tool(
@@ -1319,12 +1526,19 @@ def _benchmark_node_impl(
     - entry: 入口节点名（可选；默认用首个文件第一个 key）
     - iterations: 跑几次（1..1000，默认 10）
     - resource_path: 资源目录路径（可选，同 run_pipeline）
+    - pipeline_override: 单次 benchmark 的节点参数覆盖（可选），与 run_pipeline 同语义。
+      调参循环里用它免改文件试参数（试完确定值后再写回 pipeline JSON）：
+        benchmark_node(..., node="MyNode",
+                       pipeline_override={"MyNode": {"roi": [x, y, w, h], "threshold": 0.8}})
+      也可用 {"MyNode": {"timeout": 2000}} 压缩未命中时的单次迭代耗时（默认节点超时 20s）。
+      ⚠️ 不允许覆盖 entry / node 的 next 字段（会破坏 benchmark 的隔离链路，直接报参数错误）。
 
     实现机制：
     1. 加载完整 pipeline（用 OVERWRITE 策略，benchmark 视角下冲突无所谓）
     2. 构造 override pipeline：entry → node（强制）→ done（剥离原 next）
        避免 node 原 next 触发下游副作用
-    3. 跑 iterations 次，每次记录 wall-clock latency + node.recognition.all_results[0].score
+    3. 跑 iterations 次，每次 post_task 附带 pipeline_override（字段级合并），
+       记录 wall-clock latency + node.recognition.all_results[0].score
     4. 聚合返回 BenchmarkRunResult
 
     返回值 BenchmarkRunResult.to_dict()：
@@ -1358,7 +1572,14 @@ def benchmark_node(
     entry: Optional[str] = None,
     iterations: int = 10,
     resource_path: Optional[str] = None,
+    pipeline_override: Optional[dict] = None,
 ) -> dict | str:
     return _benchmark_node_impl(
-        controller_id, pipeline_path, node, entry, iterations, resource_path
+        controller_id,
+        pipeline_path,
+        node,
+        entry,
+        iterations,
+        resource_path,
+        pipeline_override,
     )
diff --git a/tests/test_dbg_pipeline.py b/tests/test_dbg_pipeline.py
index ec3e805..8cd1e89 100644
--- a/tests/test_dbg_pipeline.py
+++ b/tests/test_dbg_pipeline.py
@@ -3,16 +3,23 @@
 覆盖范围：
 - run_pipeline 端到端：add_resource_path → Resource override_pipeline →
   Tasker post_task → TaskDetail 解析
+- pipeline_override 端到端：post_task 级字段覆盖真实改变识别结果
+- timeout_seconds 端到端：挂死的节点被工具侧超时主动停止
 - DbgController fixture 能正常注册到 object_registry 并被 Tasker 找到
 
 依赖：tests/conftest.py 的 maa_dbg_controller fixture
 """
+import json
+import time
 from pathlib import Path
 
 import pytest
 
 from maa_mcp.pipeline_tools import run_pipeline
 
+# 兼容 fastmcp 2.x (FunctionTool 包装) 与 3.x (保持原函数)
+run_pipeline = getattr(run_pipeline, "fn", run_pipeline)  # type: ignore[arg-type]
+
 BUNDLE_FIXTURE = Path(__file__).parent / "fixtures" / "bundle_minimal"
 
 
@@ -35,3 +42,95 @@ def test_directhit_pipeline_succeeds(self, maa_dbg_controller):
         assert result["entry"] == "MyEntry"
         assert result["node_count"] >= 2  # MyEntry + MyExit
         assert result["status"] == "succeeded"
+
+
+@pytest.mark.integration
+class TestDbgControllerPipelineOverrideTimeout:
+    """pipeline_override + timeout_seconds 的端到端行为。
+
+    DbgController 返回全黑帧（见 conftest），因此：
+    - ColorMatch 找白色（lower/upper 250..255）永远不命中 → 用来构造挂死节点
+    - ColorMatch 找黑色（lower/upper 0..5）必然命中 → 用来验证 override 生效
+    """
+
+    def _write_hang_pipeline(self, tmp_path: Path) -> Path:
+        """一个在全黑帧上永远识别不命中的节点（节点级超时 60s）。"""
+        p = tmp_path / "hang.json"
+        p.write_text(
+            json.dumps(
+                {
+                    "WaitWhite": {
+                        "recognition": "ColorMatch",
+                        "lower": [250, 250, 250],
+                        "upper": [255, 255, 255],
+                        "roi": [0, 0, 64, 64],
+                        "timeout": 60000,
+                        "action": "DoNothing",
+                    }
+                }
+            ),
+            encoding="utf-8",
+        )
+        return p
+
+    def test_pipeline_override_flips_recognition_result(
+        self, maa_dbg_controller, tmp_path
+    ):
+        """文件里的节点找白色（必失败）；override 改成找黑色 → 转为成功。
+
+        证明 post_task 级 pipeline_override 做的是字段级合并，
+        且不需要改 pipeline 文件就能改变单次运行的识别参数。
+        """
+        pipeline = self._write_hang_pipeline(tmp_path)
+        result = run_pipeline(
+            controller_id=maa_dbg_controller,
+            pipeline_path=str(pipeline),
+            entry="WaitWhite",
+            resource_path=str(BUNDLE_FIXTURE),
+            start_agent=False,
+            pipeline_override={
+                "WaitWhite": {"lower": [0, 0, 0], "upper": [5, 5, 5]}
+            },
+            timeout_seconds=30,
+        )
+        assert isinstance(result, dict), f"unexpected result: {result!r}"
+        assert result["success"] is True, f"override 未生效: {result!r}"
+        assert result["status"] == "succeeded"
+
+    def test_timeout_stops_hanging_pipeline(self, maa_dbg_controller, tmp_path):
+        """不命中的节点（节点级超时 60s）被 timeout_seconds=3 主动停止。
+
+        没有工具侧超时的话，这个用例要挂满 60s 才返回。
+        """
+        pipeline = self._write_hang_pipeline(tmp_path)
+        start = time.monotonic()
+        result = run_pipeline(
+            controller_id=maa_dbg_controller,
+            pipeline_path=str(pipeline),
+            entry="WaitWhite",
+            resource_path=str(BUNDLE_FIXTURE),
+            start_agent=False,
+            timeout_seconds=3,
+        )
+        elapsed = time.monotonic() - start
+        assert isinstance(result, dict), f"unexpected result: {result!r}"
+        assert result["success"] is False
+        assert result["status"] == "timeout"
+        assert "timeout_seconds" in result["error"]
+        # 远小于节点级超时 60s（给 post_stop 清理留些余量）
+        assert elapsed < 30, f"超时停止耗时过长: {elapsed:.1f}s"
+
+    def test_harmless_override_keeps_success(self, maa_dbg_controller):
+        """对正常 pipeline 覆盖无关字段（post_delay）不影响成功结果。"""
+        result = run_pipeline(
+            controller_id=maa_dbg_controller,
+            pipeline_path=str(BUNDLE_FIXTURE / "pipeline" / "entry.json"),
+            entry="MyEntry",
+            resource_path=str(BUNDLE_FIXTURE),
+            start_agent=False,
+            pipeline_override={"MyExit": {"post_delay": 0}},
+            timeout_seconds=30,
+        )
+        assert isinstance(result, dict), f"unexpected result: {result!r}"
+        assert result["success"] is True
+        assert result["status"] == "succeeded"
diff --git a/tests/test_pipeline_override_timeout.py b/tests/test_pipeline_override_timeout.py
new file mode 100644
index 0000000..31ad573
--- /dev/null
+++ b/tests/test_pipeline_override_timeout.py
@@ -0,0 +1,611 @@
+"""run_pipeline / benchmark_node 的 pipeline_override + timeout_seconds 测试。
+
+背景：单节点验证时 run_pipeline 此前没有「只截 ROI 区域验证」的参数——
+想收紧 roi 只能改 pipeline 文件；同时 task_job.wait() 无限阻塞，
+识别未命中的节点会把 MCP 调用挂满整个节点超时（默认 20s）。
+
+覆盖范围（纯函数 + monkeypatch 假件，不依赖 maafw 运行时）：
+- _validate_timeout_seconds: 超时参数校验
+- _validate_pipeline_override: override 结构校验
+- _unknown_override_nodes: 未知节点名 warning 生成
+- _wait_task_with_timeout: 轮询等待 / 超时 post_stop 路径
+- run_pipeline: 新参数透传到 post_task、timeout 返回结构、默认值向后兼容
+- benchmark_node: override 透传、next 保护、warnings 序列化
+
+端到端（DbgController）见 tests/test_dbg_pipeline.py。
+"""
+
+import inspect
+import json
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any, Optional
+
+import pytest
+
+from maa_mcp.pipeline_tools import (
+    BenchmarkRunResult,
+    _benchmark_node_impl,
+    _unknown_override_nodes,
+    _validate_pipeline_override,
+    _validate_timeout_seconds,
+    _wait_task_with_timeout,
+    run_pipeline,
+)
+
+# 兼容 fastmcp 2.x (FunctionTool 包装) 与 3.x (保持原函数)
+run_pipeline = getattr(run_pipeline, "fn", run_pipeline)  # type: ignore[arg-type]
+
+
+# =============================================================================
+# 纯函数单元测试
+# =============================================================================
+
+
+@pytest.mark.unit
+class TestValidateTimeoutSeconds:
+    """`_validate_timeout_seconds` 参数校验。"""
+
+    def test_none_is_valid(self) -> None:
+        """None（不限时）合法，不抛错。"""
+        _validate_timeout_seconds(None)
+
+    @pytest.mark.parametrize("value", [10, 0.5, 15.0, 1])
+    def test_positive_numbers_valid(self, value) -> None:
+        """正 int / float 合法。"""
+        _validate_timeout_seconds(value)
+
+    @pytest.mark.parametrize("value", [0, -1, -0.5])
+    def test_non_positive_raises(self, value) -> None:
+        """0 和负数非法。"""
+        with pytest.raises(ValueError, match="必须 > 0"):
+            _validate_timeout_seconds(value)
+
+    @pytest.mark.parametrize("value", ["10", [10], True])
+    def test_non_numeric_raises(self, value) -> None:
+        """字符串 / list / bool 非法（bool 是 int 子类，需显式拒绝）。"""
+        with pytest.raises(ValueError, match="必须是正数"):
+            _validate_timeout_seconds(value)
+
+    @pytest.mark.parametrize(
+        "value", [float("nan"), float("inf"), float("-inf"), 10**400]
+    )
+    def test_non_finite_raises(self, value) -> None:
+        """NaN / ±inf / 超出 float 范围的巨大整数非法。
+
+        JSON 层（jiter/pydantic）接受 NaN/Infinity/1e400；不拦的话
+        deadline = monotonic() + NaN 会让超时分支永远不触发——
+        静默禁用超时还带 CPU 空转轮询。
+        """
+        with pytest.raises(ValueError, match="有限数值"):
+            _validate_timeout_seconds(value)
+
+
+@pytest.mark.unit
+class TestValidatePipelineOverride:
+    """`_validate_pipeline_override` 结构校验。"""
+
+    def test_none_returns_empty_dict(self) -> None:
+        """None → 空 dict（等价于不覆盖）。"""
+        assert _validate_pipeline_override(None) == {}
+
+    def test_valid_override_passes_through(self) -> None:
+        """合法结构原样返回。"""
+        override = {"MyNode": {"roi": [0, 0, 100, 100], "timeout": 3000}}
+        assert _validate_pipeline_override(override) is override
+
+    def test_empty_dict_valid(self) -> None:
+        """空 dict 合法（no-op 覆盖）。"""
+        assert _validate_pipeline_override({}) == {}
+
+    @pytest.mark.parametrize("value", ["not a dict", [{"A": {}}], 42])
+    def test_top_level_non_dict_raises(self, value) -> None:
+        """顶层非 dict 非法。"""
+        with pytest.raises(ValueError, match="必须是 dict"):
+            _validate_pipeline_override(value)
+
+    def test_non_string_key_raises(self) -> None:
+        """非字符串键非法。"""
+        with pytest.raises(ValueError, match="非空字符串节点名"):
+            _validate_pipeline_override({123: {"roi": [0, 0, 1, 1]}})
+
+    def test_empty_string_key_raises(self) -> None:
+        """空字符串键非法。"""
+        with pytest.raises(ValueError, match="非空字符串节点名"):
+            _validate_pipeline_override({"": {"roi": [0, 0, 1, 1]}})
+
+    def test_node_value_non_dict_raises(self) -> None:
+        """节点值非 dict 非法。"""
+        with pytest.raises(ValueError, match="必须是字段 dict"):
+            _validate_pipeline_override({"MyNode": [0, 0, 100, 100]})
+
+
+@pytest.mark.unit
+class TestUnknownOverrideNodes:
+    """`_unknown_override_nodes` warning 生成。"""
+
+    def test_all_known_no_warnings(self) -> None:
+        """节点都在 merged 中 → 无 warning。"""
+        merged = {"A": {}, "B": {}}
+        assert _unknown_override_nodes({"A": {"roi": [0, 0, 1, 1]}}, merged) == []
+
+    def test_unknown_node_warned_with_name(self) -> None:
+        """未知节点 → warning 含节点名。"""
+        warnings = _unknown_override_nodes({"Typo": {"roi": [0, 0, 1, 1]}}, {"A": {}})
+        assert len(warnings) == 1
+        assert "Typo" in warnings[0]
+
+    def test_mixed_only_unknown_warned(self) -> None:
+        """已知 + 未知混合 → 只对未知节点告警，且排序稳定。"""
+        merged = {"A": {}}
+        warnings = _unknown_override_nodes(
+            {"A": {}, "Z_Unknown": {}, "B_Unknown": {}}, merged
+        )
+        assert len(warnings) == 2
+        assert "B_Unknown" in warnings[0]
+        assert "Z_Unknown" in warnings[1]
+
+    def test_empty_override_no_warnings(self) -> None:
+        """空 override → 无 warning。"""
+        assert _unknown_override_nodes({}, {"A": {}}) == []
+
+
+# =============================================================================
+# _wait_task_with_timeout（SimpleNamespace 假件）
+# =============================================================================
+
+
+class FakeTaskJob:
+    """按 done_after 次轮询后变 done 的假 TaskJob。
+
+    done_after=0 表示立即 done；done_after=None 表示永远不 done。
+    """
+
+    def __init__(self, detail: Any, done_after: Optional[int] = 0):
+        self._detail = detail
+        self._done_after = done_after
+        self._polls = 0
+        self.wait_calls = 0
+
+    @property
+    def done(self) -> bool:
+        if self._done_after is None:
+            return False
+        self._polls += 1
+        return self._polls > self._done_after
+
+    def wait(self) -> "FakeTaskJob":
+        self.wait_calls += 1
+        return self
+
+    def get(self) -> Any:
+        return self._detail
+
+
+class FakeTasker:
+    """记录 post_stop / post_task 调用的假 Tasker。"""
+
+    def __init__(self, task_job: Optional[FakeTaskJob] = None):
+        self._task_job = task_job
+        self.post_stop_calls = 0
+        self.post_task_calls: list = []
+
+    def post_task(self, entry: str, pipeline_override: dict = {}) -> FakeTaskJob:
+        self.post_task_calls.append((entry, pipeline_override))
+        return self._task_job
+
+    def post_stop(self):
+        self.post_stop_calls += 1
+        return SimpleNamespace(wait=lambda: None)
+
+
+@pytest.mark.unit
+class TestWaitTaskWithTimeout:
+    """`_wait_task_with_timeout` 等待 / 超时行为。"""
+
+    def _detail(self) -> SimpleNamespace:
+        return SimpleNamespace(task_id=1, nodes=[])
+
+    def test_none_timeout_blocks_via_wait(self) -> None:
+        """timeout_seconds=None → 走 wait().get() 旧路径，不轮询不 post_stop。"""
+        detail = self._detail()
+        job = FakeTaskJob(detail, done_after=0)
+        tasker = FakeTasker()
+        result, timed_out = _wait_task_with_timeout(tasker, job, None)
+        assert result is detail
+        assert timed_out is False
+        assert job.wait_calls == 1
+        assert tasker.post_stop_calls == 0
+
+    def test_done_immediately_returns_detail(self) -> None:
+        """首次轮询已 done → 直接返回 detail，不 post_stop。"""
+        detail = self._detail()
+        job = FakeTaskJob(detail, done_after=0)
+        tasker = FakeTasker()
+        result, timed_out = _wait_task_with_timeout(
+            tasker, job, timeout_seconds=5, poll_interval=0.01
+        )
+        assert result is detail
+        assert timed_out is False
+        assert tasker.post_stop_calls == 0
+
+    def test_done_after_a_few_polls_returns_detail(self) -> None:
+        """几次轮询后 done → 正常返回，不超时。"""
+        detail = self._detail()
+        job = FakeTaskJob(detail, done_after=3)
+        tasker = FakeTasker()
+        result, timed_out = _wait_task_with_timeout(
+            tasker, job, timeout_seconds=5, poll_interval=0.01
+        )
+        assert result is detail
+        assert timed_out is False
+        assert tasker.post_stop_calls == 0
+
+    def test_never_done_triggers_post_stop(self) -> None:
+        """一直不 done → 超时后 post_stop 一次，返回 timed_out=True。"""
+        detail = self._detail()
+        job = FakeTaskJob(detail, done_after=None)
+        tasker = FakeTasker()
+        result, timed_out = _wait_task_with_timeout(
+            tasker, job, timeout_seconds=0.05, poll_interval=0.01
+        )
+        assert timed_out is True
+        assert tasker.post_stop_calls == 1
+        # 停止后仍然回收 detail（携带部分节点信息）
+        assert result is detail
+        # 停止后通过 wait() 取 detail（stop 完成后 wait 立即返回）
+        assert job.wait_calls == 1
+
+    def test_timeout_shorter_than_poll_interval_still_checks_done_first(self) -> None:
+        """timeout < poll_interval 时，done 检查仍先于超时判定。"""
+        detail = self._detail()
+        job = FakeTaskJob(detail, done_after=0)
+        tasker = FakeTasker()
+        result, timed_out = _wait_task_with_timeout(
+            tasker, job, timeout_seconds=0.001, poll_interval=1.0
+        )
+        assert result is detail
+        assert timed_out is False
+
+
+# =============================================================================
+# run_pipeline 新参数（monkeypatch 假 Resource / Tasker，不依赖 maafw 运行时）
+# =============================================================================
+
+
+def _write_pipeline(tmp_path: Path, name: str, nodes: dict) -> Path:
+    p = tmp_path / name
+    p.write_text(json.dumps(nodes, ensure_ascii=False), encoding="utf-8")
+    return p
+
+
+def _success_detail() -> SimpleNamespace:
+    status = SimpleNamespace(
+        succeeded=True, failed=False, running=False, pending=False, done=True
+    )
+    return SimpleNamespace(status=status, task_id=7, nodes=[])
+
+
+def _patch_resource_and_tasker(
+    monkeypatch: pytest.MonkeyPatch, tasker: FakeTasker
+) -> None:
+    """把 pipeline_tools 里的 Resource / Tasker 换成假件。"""
+    fake_resource = SimpleNamespace(override_pipeline=lambda merged: True)
+    monkeypatch.setattr(
+        "maa_mcp.pipeline_tools.get_or_create_resource", lambda: fake_resource
+    )
+    monkeypatch.setattr(
+        "maa_mcp.pipeline_tools.get_or_create_tasker", lambda cid: tasker
+    )
+
+
+@pytest.mark.unit
+class TestRunPipelineNewParamValidation:
+    """新参数的预校验：非法输入返回错误字符串，且先于任何状态修改。"""
+
+    def test_negative_timeout_returns_error(self, tmp_path: Path) -> None:
+        f = _write_pipeline(tmp_path, "p.json", {"P": {"recognition": "DirectHit"}})
+        result = run_pipeline(
+            controller_id="fake", pipeline_path=str(f), timeout_seconds=-1
+        )
+        assert isinstance(result, str)
+        assert "参数错误" in result
+        assert "timeout_seconds" in result
+
+    def test_non_dict_override_returns_error(self, tmp_path: Path) -> None:
+        f = _write_pipeline(tmp_path, "p.json", {"P": {"recognition": "DirectHit"}})
+        result = run_pipeline(
+            controller_id="fake",
+            pipeline_path=str(f),
+            pipeline_override="not a dict",
+        )
+        assert isinstance(result, str)
+        assert "参数错误" in result
+        assert "pipeline_override" in result
+
+    def test_defaults_are_backward_compatible(self) -> None:
+        """新参数默认值必须是 None（不改变旧行为）。"""
+        sig = inspect.signature(run_pipeline)
+        assert sig.parameters["pipeline_override"].default is None
+        assert sig.parameters["timeout_seconds"].default is None
+
+
+@pytest.mark.unit
+class TestRunPipelineOverridePassthrough:
+    """pipeline_override 透传到 tasker.post_task。"""
+
+    def test_override_reaches_post_task(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """覆盖 dict 原样传给 post_task 第二参数。"""
+        f = _write_pipeline(
+            tmp_path, "p.json", {"MyNode": {"recognition": "DirectHit"}}
+        )
+        job = FakeTaskJob(_success_detail(), done_after=0)
+        tasker = FakeTasker(job)
+        _patch_resource_and_tasker(monkeypatch, tasker)
+
+        override = {"MyNode": {"roi": [10, 20, 300, 100], "timeout": 3000}}
+        result = run_pipeline(
+            controller_id="fake",
+            pipeline_path=str(f),
+            start_agent=False,
+            pipeline_override=override,
+        )
+        assert isinstance(result, dict), f"unexpected: {result!r}"
+        assert result["success"] is True
+        assert tasker.post_task_calls == [("MyNode", override)]
+        # 已知节点：无 warnings
+        assert "warnings" not in result
+
+    def test_no_override_posts_empty_dict(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """不传 override → post_task 收到空 dict（与 maafw 默认一致）。"""
+        f = _write_pipeline(
+            tmp_path, "p.json", {"MyNode": {"recognition": "DirectHit"}}
+        )
+        job = FakeTaskJob(_success_detail(), done_after=0)
+        tasker = FakeTasker(job)
+        _patch_resource_and_tasker(monkeypatch, tasker)
+
+        result = run_pipeline(
+            controller_id="fake", pipeline_path=str(f), start_agent=False
+        )
+        assert isinstance(result, dict), f"unexpected: {result!r}"
+        assert tasker.post_task_calls == [("MyNode", {})]
+
+    def test_unknown_override_node_warned(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """覆盖了不在文件节点表中的节点名 → warnings 提示（不阻断执行）。"""
+        f = _write_pipeline(
+            tmp_path, "p.json", {"MyNode": {"recognition": "DirectHit"}}
+        )
+        job = FakeTaskJob(_success_detail(), done_after=0)
+        tasker = FakeTasker(job)
+        _patch_resource_and_tasker(monkeypatch, tasker)
+
+        result = run_pipeline(
+            controller_id="fake",
+            pipeline_path=str(f),
+            start_agent=False,
+            pipeline_override={"TypoNode": {"roi": [0, 0, 1, 1]}},
+        )
+        assert isinstance(result, dict), f"unexpected: {result!r}"
+        assert result["success"] is True  # 未知名只警告，不失败
+        assert any("TypoNode" in w for w in result.get("warnings", []))
+
+
+@pytest.mark.unit
+class TestRunPipelineTimeout:
+    """timeout_seconds 超时路径的返回结构。"""
+
+    def test_timeout_returns_structured_result(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """任务一直不结束 → status='timeout'，success=False，tasker 被 post_stop。"""
+        f = _write_pipeline(
+            tmp_path, "p.json", {"MyNode": {"recognition": "DirectHit"}}
+        )
+        # 超时后回收的部分 detail（status 无所谓，nodes 带一个已执行节点）
+        partial_detail = SimpleNamespace(
+            status=SimpleNamespace(
+                succeeded=False, failed=True, running=False, pending=False, done=True
+            ),
+            task_id=9,
+            nodes=[SimpleNamespace(name="MyNode", recognition=None)],
+        )
+        job = FakeTaskJob(partial_detail, done_after=None)
+        tasker = FakeTasker(job)
+        _patch_resource_and_tasker(monkeypatch, tasker)
+
+        result = run_pipeline(
+            controller_id="fake",
+            pipeline_path=str(f),
+            start_agent=False,
+            timeout_seconds=0.05,
+        )
+        assert isinstance(result, dict), f"unexpected: {result!r}"
+        assert result["success"] is False
+        assert result["status"] == "timeout"
+        assert "timeout_seconds" in result["error"]
+        assert tasker.post_stop_calls == 1
+        # 超时前已执行的部分节点详情要带回来
+        assert result["nodes"] == [{"name": "MyNode"}]
+        assert result["entry"] == "MyNode"
+        assert result["task_id"] == 9
+
+    def test_fast_task_within_timeout_unaffected(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """任务在限时内完成 → 与不限时结果一致。"""
+        f = _write_pipeline(
+            tmp_path, "p.json", {"MyNode": {"recognition": "DirectHit"}}
+        )
+        job = FakeTaskJob(_success_detail(), done_after=0)
+        tasker = FakeTasker(job)
+        _patch_resource_and_tasker(monkeypatch, tasker)
+
+        result = run_pipeline(
+            controller_id="fake",
+            pipeline_path=str(f),
+            start_agent=False,
+            timeout_seconds=30,
+        )
+        assert isinstance(result, dict), f"unexpected: {result!r}"
+        assert result["success"] is True
+        assert result["status"] == "succeeded"
+        assert tasker.post_stop_calls == 0
+
+
+# =============================================================================
+# benchmark_node 的 pipeline_override
+# =============================================================================
+
+
+@pytest.mark.unit
+class TestBenchmarkNodePipelineOverride:
+    """`_benchmark_node_impl` 的 override 透传与保护。"""
+
+    def _pipeline(self, tmp_path: Path) -> Path:
+        return _write_pipeline(
+            tmp_path,
+            "bench.json",
+            {
+                "Entry": {"recognition": "DirectHit", "next": ["Target"]},
+                "Target": {"recognition": "OCR", "expected": "X"},
+            },
+        )
+
+    def test_invalid_override_structure_returns_error(self, tmp_path: Path) -> None:
+        result = _benchmark_node_impl(
+            controller_id="fake",
+            pipeline_path=str(self._pipeline(tmp_path)),
+            node="Target",
+            pipeline_override="oops",
+        )
+        assert isinstance(result, str)
+        assert "参数错误" in result
+
+    @pytest.mark.parametrize("protected", ["Entry", "Target"])
+    def test_override_next_on_harness_nodes_rejected(
+        self, tmp_path: Path, protected: str
+    ) -> None:
+        """覆盖 entry / node 的 next → 参数错误（保护隔离链路）。"""
+        result = _benchmark_node_impl(
+            controller_id="fake",
+            pipeline_path=str(self._pipeline(tmp_path)),
+            node="Target",
+            pipeline_override={protected: {"next": ["Somewhere"]}},
+        )
+        assert isinstance(result, str)
+        assert "参数错误" in result
+        assert "next" in result
+
+    def test_override_reaches_each_post_task(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """每次迭代的 post_task 都带上 override。"""
+        detail = SimpleNamespace(
+            status=SimpleNamespace(
+                succeeded=True, failed=False, running=False, pending=False, done=True
+            ),
+            task_id=1,
+            nodes=[],
+        )
+
+        class MultiJobTasker(FakeTasker):
+            def post_task(self, entry, pipeline_override={}):
+                self.post_task_calls.append((entry, pipeline_override))
+                return FakeTaskJob(detail, done_after=0)
+
+        tasker = MultiJobTasker()
+        _patch_resource_and_tasker(monkeypatch, tasker)
+
+        override = {"Target": {"roi": [0, 0, 50, 50], "timeout": 2000}}
+        result = _benchmark_node_impl(
+            controller_id="fake",
+            pipeline_path=str(self._pipeline(tmp_path)),
+            node="Target",
+            iterations=3,
+            pipeline_override=override,
+        )
+        assert isinstance(result, dict), f"unexpected: {result!r}"
+        assert len(tasker.post_task_calls) == 3
+        assert all(call == ("Entry", override) for call in tasker.post_task_calls)
+
+    def test_unknown_override_node_in_warnings(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """未知节点名 → 返回值 warnings 提示。"""
+        detail = SimpleNamespace(
+            status=SimpleNamespace(
+                succeeded=True, failed=False, running=False, pending=False, done=True
+            ),
+            task_id=1,
+            nodes=[],
+        )
+
+        class MultiJobTasker(FakeTasker):
+            def post_task(self, entry, pipeline_override={}):
+                self.post_task_calls.append((entry, pipeline_override))
+                return FakeTaskJob(detail, done_after=0)
+
+        tasker = MultiJobTasker()
+        _patch_resource_and_tasker(monkeypatch, tasker)
+
+        result = _benchmark_node_impl(
+            controller_id="fake",
+            pipeline_path=str(self._pipeline(tmp_path)),
+            node="Target",
+            iterations=1,
+            pipeline_override={"TypoNode": {"roi": [0, 0, 1, 1]}},
+        )
+        assert isinstance(result, dict), f"unexpected: {result!r}"
+        assert any("TypoNode" in w for w in result.get("warnings", []))
+
+
+@pytest.mark.unit
+class TestBenchmarkRunResultWarnings:
+    """BenchmarkRunResult.warnings 序列化行为。"""
+
+    def _result(self, warnings) -> BenchmarkRunResult:
+        return BenchmarkRunResult(
+            node="Target",
+            iterations=1,
+            successes=1,
+            min_score=0.9,
+            max_score=0.9,
+            mean_score=0.9,
+            latency_ms=[100],
+            all_results_samples=[],
+            warnings=warnings,
+        )
+
+    def test_empty_warnings_omitted(self) -> None:
+        """空 warnings 不序列化（保持返回体精简，与 PipelineLoadResult 一致）。"""
+        assert "warnings" not in self._result([]).to_dict()
+
+    def test_non_empty_warnings_serialized_as_new_list(self) -> None:
+        """非空 warnings 序列化，且返回新 list。"""
+        r = self._result(["w1"])
+        d = r.to_dict()
+        assert d["warnings"] == ["w1"]
+        d["warnings"].append("w2")
+        assert r.warnings == ["w1"]
+
+    def test_default_warnings_empty(self) -> None:
+        """不传 warnings 时默认空 list（向后兼容旧构造方式）。"""
+        r = BenchmarkRunResult(
+            node="T",
+            iterations=1,
+            successes=0,
+            min_score=None,
+            max_score=None,
+            mean_score=None,
+            latency_ms=[10],
+            all_results_samples=[],
+        )
+        assert r.warnings == []
+        assert "warnings" not in r.to_dict()