Emit a Lean verification file for `ir`.
(ir: WorkflowIR, prefix: str)
| 1017 | # ============================================================================ |
| 1018 | |
| 1019 | def generate_lean(ir: WorkflowIR, prefix: str) -> str: |
| 1020 | """Emit a Lean verification file for `ir`.""" |
| 1021 | lines: list[str] = [] |
| 1022 | |
| 1023 | # Header |
| 1024 | lines.append("import Lean") |
| 1025 | lines.append("import Mathlib") |
| 1026 | lines.append("import AgentVerifier.Basics") |
| 1027 | lines.append("") |
| 1028 | lines.append("namespace AgenticKernel") |
| 1029 | lines.append("") |
| 1030 | |
| 1031 | # Banner |
| 1032 | lines.append("/-") |
| 1033 | lines.append("=" * 80) |
| 1034 | lines.append(f"STATIC VERIFICATION: {ir.name}") |
| 1035 | lines.append(f"Goal: {ir.goal}") |
| 1036 | lines.append(f"Parameters: {[p.name for p in ir.parameters]}") |
| 1037 | lines.append(f"Nodes: {len(ir.nodes)}, Entry: {ir.entry}, Exits: {ir.exits}") |
| 1038 | lines.append("") |
| 1039 | for n in ir.nodes: |
| 1040 | rstr = ", ".join(r.name for r in n.reads) or "(none)" |
| 1041 | wstr = ", ".join(w.name for w in n.writes) or "(none)" |
| 1042 | tag = "[DET]" if not n.is_llm_node else "[LLM]" |
| 1043 | lines.append(f" Node {n.id:3d}: {n.step_type:18s} {tag:6s} \"{n.name}\"") |
| 1044 | lines.append(f" reads: {rstr}") |
| 1045 | lines.append(f" writes: {wstr}") |
| 1046 | lines.append("=" * 80) |
| 1047 | lines.append("-/") |
| 1048 | lines.append("") |
| 1049 | |
| 1050 | # STEP 1: WORKFLOW GRAPH |
| 1051 | lines.append("/-") |
| 1052 | lines.append("=" * 72) |
| 1053 | lines.append("STEP 1: WORKFLOW GRAPH") |
| 1054 | lines.append("=" * 72) |
| 1055 | lines.append("-/") |
| 1056 | lines.append("") |
| 1057 | lines.append("-- Node IDs") |
| 1058 | for n in ir.nodes: |
| 1059 | lines.append(f"def {prefix}_nodeId{n.id} : NodeId := ⟨{n.id}⟩") |
| 1060 | lines.append("") |
| 1061 | |
| 1062 | for n in ir.nodes: |
| 1063 | ri = ", ".join(v.to_lean_typed() for v in n.reads) |
| 1064 | wi = ", ".join(v.to_lean_typed() for v in n.writes) |
| 1065 | nm = f'some "{n.name}"' if n.name else "none" |
| 1066 | if n.instruction: |
| 1067 | esc = n.instruction.replace('\\', '\\\\').replace('"', '\\"').replace('\n', '\\n') |
| 1068 | ins = f'some "{esc}"' |
| 1069 | else: |
| 1070 | ins = "none" |
| 1071 | lines.append(f"-- Node {n.id}: {n.step_type} \"{n.name}\"") |
| 1072 | lines.append(f"def {prefix}_node{n.id} : WorkflowNode := {{") |
| 1073 | lines.append(f" id := {prefix}_nodeId{n.id}, name := {nm}") |
| 1074 | lines.append(f" stepType := .{n.step_type}") |
| 1075 | lines.append(f" reads := [{ri}], writes := [{wi}]") |
| 1076 | lines.append(f" llmInstruction := {ins}") |
no test coverage detected