win_monitor/backend/app/automation/nodes/human.py

from __future__ import annotations

from typing import Any

from ..context import WorkflowContext, WorkflowPaused
from ..registry import control_ports, field_def, register_node


def ask_user_node(node: dict[str, Any], inputs: dict[str, Any], context: WorkflowContext) -> dict[str, Any]:
    question = str(inputs.get("question", node.get("params", {}).get("question")) or "请确认下一步操作。")
    raise WorkflowPaused(
        question,
        {
            "node_id": node.get("id"),
            "question": question,
            "screenshot_path": context.runtime.get("current_screenshot_path"),
        },
    )


register_node(
    {
        "type": "human.ask_user",
        "category": "human",
        "label": "询问用户",
        "params": {"question": field_def("textarea", "问题", "请确认下一步操作。")},
        "inputs": {"question": field_def("string", "问题")},
        "outputs": {"answer": {"type": "string", "label": "用户回答"}},
        "control_ports": control_ports(["answered", "failure"]),
    },
    ask_user_node,
)