Class NodeIR

verifier/AgentVerifier/WorkflowToLean.py:84–134

A workflow node.

@dataclass
class NodeIR:
    """A workflow node."""
    id: int
    name: str
    step_type: str                                  # one of the 13 Lean StepType names
    reads: list[TypedVar]
    writes: list[TypedVar]
    instruction: Optional[str] = None
    submodule_ref: Optional[SubmoduleRef] = None

    @property
    def exec_type(self) -> str:
        # Mirror AgentVerifier/YamlStepType.lean::StepType.execType
        deterministic = {
            "forEachLoop", "whileLoop", "conditional", "switchBranch",
            "setVariable", "incrementVariable", "returnValue", "input",
            "parallel", "gather",
        }
        composition = {"call"}
        if self.step_type in deterministic:
            return "deterministic"
        if self.step_type in composition:
            return "composition"
        return "unstructured"

    @property
    def is_llm_node(self) -> bool:
        """True iff the node's behaviour depends on an LLM call."""
        return self.exec_type != "deterministic"

    def to_dict(self) -> dict:
        d: dict = {
            "id": self.id, "name": self.name, "step_type": self.step_type,
            "reads": [r.to_dict() for r in self.reads],
            "writes": [w.to_dict() for w in self.writes],
        }
        if self.instruction is not None:
            d["instruction"] = self.instruction
        if self.submodule_ref is not None:
            d["submodule_ref"] = self.submodule_ref.to_dict()
        return d

    @staticmethod
    def from_dict(d: dict) -> "NodeIR":
        return NodeIR(
            id=d["id"], name=d["name"], step_type=d["step_type"],
            reads=[TypedVar.from_dict(r) for r in d["reads"]],
            writes=[TypedVar.from_dict(w) for w in d["writes"]],
            instruction=d.get("instruction"),
            submodule_ref=SubmoduleRef.from_dict(d["submodule_ref"]) if d.get("submodule_ref") else None,
        )


# ---------- EdgeIR ----------