pipeline-testing

name	pipeline-testing
description	Test and validate MaaFramework Pipeline JSON, recognition nodes, action nodes, CustomAction/CustomRecognition wiring, resource loading, and end-to-end task behavior. Use when running `run_pipeline`, checking OCR/TemplateMatch/ColorMatch ROI, validating `context.run_task()` results, auditing `action: Custom` or `recognition: Custom` registrations, or choosing the right resource-check command.

Pipeline Testing Skill

概述

测试 MaaFramework Pipeline JSON 中的 node，验证识别和操作是否正常工作。

历史审查后的测试边界

• 资源加载检查是 smoke test，不是端到端证明：python tools/ci/check_resource.py assets/resource/base 能证明资源包可加载，但不能证明 CustomAction/CustomRecognition 名称存在、参数路径正确、run_task() 分支判断正确。
• Custom 映射必须单独查：所有 action: Custom / action.type = Custom 的 custom_action 都要能映射到 @AgentServer.custom_action(...)；所有 recognition.type = Custom 的 custom_recognition 都要能映射到 @AgentServer.custom_recognition(...)。
• CustomRecognition 也要测：M9A 证明复杂 OCR/list/image 逻辑经常放在 CustomRecognition，不要只测 CustomAction。
• 高风险链路要复测稳定态：购买、消耗、战斗开始、继续挑战、结算确认等动作后，先识别稳定页面或完成态，再继续危险动作。
• 关闭节点不能继续调用：当 option 把可执行节点的 enable/enabled 设为 false 时，Python 必须先短路，不能再调用该节点的 run_recognition()/run_task()；同时验证关闭可选分支后，同级战斗/调查分支仍会继续识别。

Custom 映射快速检查

rg -n '"Custom"|custom_action|custom_recognition' assets/resource -g '*.json'
rg -n '@AgentServer\.custom_action|@AgentServer\.custom_recognition' agent -g '*.py'

人工对照时按名字精确匹配。资源检查通过但这里缺注册，运行时仍会失败。

核心流程

# 1. 连接设备
find_adb_device_list()
controller_id = connect_adb_device(device_name="xxx")
# 或
controller_id = connect_window(window_name="xxx")

# 2. 加载 pipeline
load_pipeline(pipeline_path="<pipeline_json>")
RESOURCE_PATH = "<resource_base_path>"

# 3. 逐个测试
for node_name in pipeline_nodes:
    result = run_pipeline(
        controller_id=controller_id,
        pipeline_path=pipeline_path,
        entry=node_name,
        resource_path=RESOURCE_PATH
    )
    # succeeded + all_results 有内容 = 识别成功
    # 详见下方"结果判断"小节

结果判断

status	all_results	含义
succeeded	有内容	✅ 识别成功
succeeded	空	❌ 识别失败
failed	—	❌ 节点未触发 / 识别超时

返回结构：

{
  "status": "succeeded",
  "nodes": [{
    "name": "node_name",
    "recognition": {
      "all_results": [{"box": [x, y, w, h], "score": 0.99, "text": "..."}]
    }
  }]
}

run_task() 结果判断 ⚠️

MCP 的 run_pipeline 单节点结果和 Python CustomAction 里的 context.run_task() 结果不是同一个使用场景。context.run_task() 的 result.nodes 可能包含:

• 真正命中并执行完成的节点
• next 列表里被尝试过但失败的节点
• 调试面板显示红叉的节点

因此不要写:

# ❌ 错：有 nodes 不代表目标节点命中
result = context.run_task("AutoSky_CloneDied")
if result and result.nodes:
    handle_clone_loss()

应该检查节点是否真的完成或识别命中:

def task_result_has_hit(result, names: set[str]) -> bool:
    if not result or not result.nodes:
        return False
    for node in result.nodes:
        if getattr(node, "name", None) not in names:
            continue
        if getattr(node, "completed", False):
            return True
        recognition = getattr(node, "recognition", None)
        if recognition and getattr(recognition, "hit", False):
            return True
    return False

实战信号：调试面板里目标节点是红叉，但日志却进入了成功分支，基本就是把"节点出现过"误当成"节点命中"。

对于会回到稳定页面的流程，建议先检测稳定状态:

if task_result_has_hit(result, {"AutoSky_CheckExplorationInfo"}):
    return "done"

再跑危险兜底动作（取消结算、购买确认、继续挑战等），避免页面已经回稳后误触发。

跨测试导航 ⚠️

Click 节点会让页面跳转，破坏后续测试初始状态。必须用 BackButton_500ms 返回：

# 测试节点
run_pipeline(..., entry="NodeA", ...)

# 必须返回！不要 click_key(4) 模拟 ESC（可能进错页面）
run_pipeline(..., entry="BackButton_500ms", ...)

# 测下一个
run_pipeline(..., entry="NodeB", ...)

BackButton_500ms 在 main_ui.json 里，DirectHit 识别返回箭头，定位精确可靠。

ROI Sweep 测试方法

OCR 失败时不要换 TemplateMatch！先用 sweep 找 sweet spot。

操作流程：

1. 生成测试 pipeline — 同一节点，多个 expand 变体
2. 关键：expected 必须匹配当前资源实际显示文本；在 MaaGumballs 中文资源里是 ["角色"]，不是英文 key ["Role"]
3. run_pipeline 逐个跑，记录 score
4. 选 score ≥ 0.99 且 ROI 不重叠相邻元素的 expand
5. 用 generate_node.py --expand X --overwrite 正式写入

实战矩阵（5 节点实测）：

节点	e0	e20	e50	最佳
角色	✅	✅ 0.9997	✅	e20
编队	✅	✅ 0.998	✅	e20
城堡	✅	❌	—	e3（仅 0-15）
佣兵团	❌	✅	✅	e20
前往群岛	✅	✅	✅	e20

观察：

• OCR 引擎非确定性：同一 ROI 不同次测试结果可能不同
• timeout: 2000 期间 OCR 会重试多次，所以单次失败不一定是真失败
• 特殊节点（如"城堡"）需小 ROI，因为大 ROI 包含邻近 UI 元素会干扰 OCR

资源保护 ⚠️

测试时绝对不要点击这些按钮（消耗资源）：

• 升级建筑、神殿升级
• 供奉、祭拜先祖
• 购买物品、商城购买
• 确认战斗开始
• 任何有资源消耗的确认按钮

误入后处理：尝试 BackButton_500ms 或 ESC → 切换 tab 刷新。

常见问题

现象	排查
节点超时失败	截屏看实际界面 → 调整 expected 文本或 roi
OCR 拆分多字（如"角色"→"电"+"色"）	缩小 ROI 避开干扰
误识别相邻文字	缩小 ROI 限制范围
点击位置不准	实际点击取 box 中心点：x + w/2, y + h/2
可滚动 UI 单视图测不全	一个截图里只能看到 5-6 个元素，需要 swipe 多次分别测；详见下方
Click + BackButton 跳错页面	Click 子页后按 BackButton 可能回到大地图/外层 UI，不是预期父页面；需主动 navigate 回来
部分可见元素识别	边界处只露 25px 也能 OCR 找到（score 偏低 0.8x），用大 ROI 覆盖即可

可滚动 UI 多视图测试

单次截图只能测当前可见的 5-6 个元素。要测全列表（如城堡 10 个建筑）必须多次 swipe：

1. 进入目标页（如城堡顶部）
2. run_pipeline 测当前可见的 N 个节点
3. 每个 click 后 BackButton 返回
4. swipe 滚动一屏
5. 再 run_pipeline 测新可见的节点
6. 重复直到列表测完

典型案例：城堡建筑（10 节点）

• 顶部：城堡管理/市场/铁匠铺/炼金工坊/训练所
• 中段：城堡主厅/神殿/家族
• 底部：藏馆/庄园

关键：所有节点共用同一大 ROI [x, top_y, w, full_h]，不要给每个节点算不同 ROI（滚动后位置变了，原 ROI 失效）。

卡住时截图查看

症状：节点超时、OCR 找不到、行为异常、识别 score 突然下降

解决：调 screencap(controller_id) 看实际屏幕状态：

• 当前在哪个页面？（是否已跳到其他页面）
• 目标文字是否被遮挡/弹窗挡住？
• 屏幕是否在加载/转圈？
• 滚动位置对吗？

配合 ocr() 看识别结果交叉验证。不要盲调 ROI/expected，先看实际屏幕。

跨测试导航 ⚠️（Click 节点路径风险）

Click 节点会让页面跳转，破坏后续测试初始状态。必须用 BackButton_500ms 返回：

# 测试节点
run_pipeline(..., entry="NodeA", ...)

# 必须返回！不要 click_key(4) 模拟 ESC（可能进错页面）
run_pipeline(..., entry="BackButton_500ms", ...)

注意：BackButton 不一定回到你期望的父页面！

• 例：城堡管理 → BackButton → 实际回到 大地图（不是城堡）
• 例：训练所 → BackButton → 回到城堡（正常）

测试前先确认 BackButton 路径，或者用 ocr() 检查当前页面状态再继续。

⚠️ MCP start_agent=true 失败时的 Fallback 方案

症状:run_pipeline(..., start_agent=true) 报

Agent 启动失败: Agent 启动失败 (identifier=12345): connect() 返回 False

根因(常见):

1. MaaFramework 的 agent 子进程(socket 进程)未在预期时间窗口内完成 MaaAgentServerStartUp
2. Socket 路径冲突 / 权限问题
3. MaaFramework 内部 MaaAgentServerNotImpl(这个错误明确说"用 MaaFramework,不要用 MaaAgentServer")

Fallback 1:单节点测试(推荐):

用 start_agent=false + 指定 entry 测单个节点(只测 OCR/Template 识别,跳过 CustomAction 链):

# 不开 agent,直接测某个节点
result = run_pipeline(
    controller_id=cid,
    pipeline_path="f:/workspace/MaaGumballs/assets/resource/base/pipeline/sky.json",
    entry="AutoSky_BagConfig",   # 任意节点名
    resource_path="f:/workspace/MaaGumballs/assets/resource/base",
    start_agent=False,          # ← 关键:false
)

适用场景:测 OCR / TemplateMatch / ColorMatch 等识别节点的 ROI 和 expected 文本是否正确。

Fallback 2:手动启动 agent 后 MCP 通过 socket 连接:

如果单节点测试不够(需测完整任务链),手动跑 agent 进程,然后用 connect_adb_device / connect_window 让 MCP 复用。

# 步骤 1:手动启动 agent
import subprocess
agent_proc = subprocess.Popen(
    ["python", "agent/main.py", "test_socket"],
    cwd=".",
)

# 步骤 2:让 MCP 连接(需要 MCP 内部 socket ID 匹配)
# 实际中,MCP 启动 agent 会自动起 socket,如果手动起需要 MCP 端配对
# 通常这一步是 MCP 内部行为,我们无法直接控制

Fallback 3:放弃 MCP,直接用 MaaFramework Python API:

import sys
sys.path.insert(0, "f:/workspace/MaaGumballs")
from maa.agent.agent_server import AgentServer
from maa.toolkit import Toolkit
from maa.resource import Resource
from maa.controller import AdbController
from maa.tasker import Tasker

Toolkit.init_option("./")
res = Resource(); res.post_bundle("./assets/resource/base").wait()
ctrl = AdbController(adb_path="adb", address="127.0.0.1:16384"); ctrl.post_connection().wait()
tasker = Tasker(); tasker.bind(res, ctrl)

import action.sky
autosky = action.sky.AutoSky()
result = autosky.run(tasker.context, type('X', (), {'custom_action_param': '{}'})())

何时用哪种:

• Fallback 1(单节点):调试 ROI / expected / TemplateMatch,快速迭代 → 首选
• Fallback 3(直接 API):测完整 task 链,真实环境验证 → MCP 不可用时用
• Fallback 2(手动起 agent):通常不需要,MCP 会自动处理

测试模板

## <文件名> 测试记录

日期: 2026-06-05
设备: xxx
controller_id: xxx

### Node 列表
- [ ] node_name_1
- [ ] node_name_2

### 测试进度
| Node | 结果 | 备注 |
|------|------|------|
| xxx  | ✅   | score=0.999 |
| xxx  | ❌   | 原因xxx |

Reference（详见 pipeline-guide）

• ROI / box / target 概念
• 识别类型：DirectHit / OCR / TemplateMatch / FeatureMatch / ColorMatch / And / Or
• 动作类型：Click / LongPress / Swipe / Scroll / InputText / ClickKey
• 节点生命周期：pre_wait_freezes → pre_delay → action → post_wait_freezes → post_delay → 截图识别 next