Function parse_task_json

verifier/AgentVerifier/WorkflowToLean.py:388–1012

Parse parsed_task.json → WorkflowIR.

(json_path: str)

def parse_task_json(json_path: str) -> WorkflowIR:
    """Parse parsed_task.json → WorkflowIR."""
    with open(json_path) as f:
        data = json.load(f)

    wf_name = data["name"]
    wf_goal = data.get("goal", "")

    # ---- Parameters ----
    params: list[TypedVar] = []
    raw_params = data.get("parameters", {}) or {}
    for pname, pval in raw_params.items():
        pval = _resolve_env_vars(pval)
        ptype = _infer_param_type(pval)
        params.append(TypedVar(pname, ptype, value=pval))

    type_ctx: dict[str, str] = {p.name: p.base_type for p in params}
    nodes: list[NodeIR] = []
    edges: list[EdgeIR] = []
    nid = 0
    loop_patches: list[dict] = []

    def alloc_id() -> int:
        nonlocal nid
        cur = nid
        nid += 1
        return cur

    def resolve_type(v: str) -> str:
        return type_ctx.get(v, "TString")

    def build_reads(vs: list[str]) -> list[TypedVar]:
        return [TypedVar(v, resolve_type(v)) for v in vs]

    def add_seq(f: Optional[int], t: int):
        if f is not None:
            edges.append(EdgeIR("seq", from_node=f, to_node=t))

    # ---------- Recursive body-step parser ----------
    def parse_body_steps(body_steps: list, body_prev: Optional[int]) -> tuple:
        bf: Optional[int] = None
        extra_tails: list[int] = []

        def link_to(new_id: int) -> None:
            nonlocal extra_tails
            add_seq(body_prev, new_id)
            if extra_tails:
                for tail_id in list(dict.fromkeys(extra_tails)):
                    add_seq(tail_id, new_id)
                extra_tails = []

        for bs in body_steps:
            bk = list(bs.keys())[0]
            bd = bs[bk]

            if bk in ("step", "task", "synthesize", "discover"):
                bid = alloc_id()
                bname = bd.get("name", f"{bk}_{bid}")