Function check_node

workflow/nodes.py:153–185

(
    state: WorkflowState,
    *,
    context: WorkflowContext,
    sender_name: str = "Check",
)

def check_node(
    state: WorkflowState,
    *,
    context: WorkflowContext,
    sender_name: str = "Check",
) -> dict[str, Any]:
    decision, check_count = classify_check_result(
        target=context.target or context.problem,
        history=context.history if context.history else state["message"],
        check_count=state["check_count"],
        benchmark_name=context.benchmark_name,
        remaining_vulns=len(state["vulns"]),
    )

    if decision == "success":
        check_message = (
            f"Successfully exploited the vulnerability, a total of {check_count} steps were attempted"
        )
    elif decision == "retry":
        check_message = f"Failed to exploit the vulnerability, please try again. {context.problem}"
    elif decision == "another":
        check_message = "Failed to exploit the vulnerability, please try another vulnerability."
    else:
        check_message = "Failed to exploit the vulnerability."

    context.history.append(f"{sender_name}{check_count}_{check_message}")

    return {
        "message": [_human_message(check_message)],
        "sender": sender_name,
        "vulns": state["vulns"],
        "check_count": check_count,
    }


@dataclass(slots=True)