""" TD Lang Executor — Runs compiled .td scripts and tracks lineage. Two modes: - compile: Parse .td -> generate .py file (no execution) - run: Parse .td -> generate .py -> execute it All outputs go to td_lang_outputs/_/ - compiled.py — The generated Python script - lineage.json — What happened, in what order (artifact tracking) Pipeline: .td file -> Parser -> AST -> Compiler -> Python string -> **Executor** """ import ast as python_ast import hashlib import json import subprocess import sys from datetime import datetime from pathlib import Path from typing import Optional from .grammar import parse_td_file, parse_td_string from .compiler import compile_program from .ast_nodes import TDProgram from .errors import TDCompileError, TDLangError # ============================================================================ # EXECUTOR # ============================================================================ class TDExecutor: """Execute td_lang programs — compile and optionally run. Usage: executor = TDExecutor() # Compile only (check + generate .py) py_path = executor.compile("demo.td") # Compile and run result = executor.run("demo.td") # Just check syntax executor.check("demo.td") """ def __init__(self, output_dir: str = "td_lang_outputs"): self.output_dir = Path(output_dir) def check(self, td_path: str) -> TDProgram: """Parse and validate a .td file without compiling or running. Args: td_path: Path to the .td file. Returns: The parsed TDProgram. Raises: TDSyntaxError: If syntax is invalid. TDCompileError: If semantic validation fails. """ print(f"[td_lang] Checking {td_path}...") program = parse_td_file(td_path) # Count what we found n_commands = len(program.commands) has_gates = program.gates is not None has_budget = program.budget is not None print(f"[td_lang] OK — {n_commands} commands", end="") if has_gates: print(f", gates: {program.gates.must_pass}", end="") if has_budget: print(f", budget set", end="") print() return program def compile(self, td_path: str) -> Path: """Parse, validate, and compile a .td file into Python. Args: td_path: Path to the .td file. Returns: Path to the generated .py file. Raises: TDSyntaxError: If syntax is invalid. TDCompileError: If compilation fails. """ print(f"[td_lang] Compiling {td_path}...") # Parse program = parse_td_file(td_path) # Compile python_code = compile_program(program) # Validate the generated Python is valid syntax try: python_ast.parse(python_code) except SyntaxError as e: raise TDCompileError( f"Generated Python has a syntax error (this is a td_lang bug): {e}", hint="Please report this — the compiler generated bad code.", ) from e # Save to output directory out_dir = self._make_output_dir(td_path) py_path = out_dir / "compiled.py" py_path.write_text(python_code) # Save source hash for lineage source_text = Path(td_path).read_text() meta = { "source_file": str(td_path), "source_hash": hashlib.sha256(source_text.encode()).hexdigest(), "compiled_at": datetime.now().isoformat(), "td_lang_version": "0.2.0", "python_file": str(py_path), "n_commands": len(program.commands), "has_gates": program.gates is not None, "has_budget": program.budget is not None, } meta_path = out_dir / "compile_meta.json" meta_path.write_text(json.dumps(meta, indent=2)) print(f"[td_lang] Compiled to: {py_path}") return py_path def run(self, td_path: str, dry_run: bool = False) -> dict: """Parse, compile, and execute a .td file. Args: td_path: Path to the .td file. dry_run: If True, compile but don't execute. Returns: Dict with execution results. """ # Compile first py_path = self.compile(td_path) if dry_run: print("[td_lang] Dry run — compiled but not executed.") return {"status": "dry_run", "compiled": str(py_path)} # Execute the generated Python script print(f"[td_lang] Executing {py_path}...") print() try: result = subprocess.run( [sys.executable, str(py_path.resolve())], capture_output=False, # Let output stream to console # Run from the original working directory, NOT the output dir. # The compiled code uses relative paths like "td_lang_outputs/..." # which must resolve from the project root. ) if result.returncode == 0: print() print("[td_lang] Execution completed successfully.") return {"status": "success", "compiled": str(py_path)} else: print() print(f"[td_lang] Execution failed (exit code {result.returncode}).") return { "status": "failed", "compiled": str(py_path), "exit_code": result.returncode, } except Exception as e: print(f"\n[td_lang] Execution error: {e}") return {"status": "error", "compiled": str(py_path), "error": str(e)} def _make_output_dir(self, td_path: str) -> Path: """Create a timestamped output directory for this run.""" name = Path(td_path).stem timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") out_dir = self.output_dir / f"{name}_{timestamp}" out_dir.mkdir(parents=True, exist_ok=True) return out_dir # ============================================================================ # PUBLIC API # ============================================================================ def check_td_file(td_path: str) -> TDProgram: """Quick syntax check on a .td file.""" return TDExecutor().check(td_path) def compile_td_file(td_path: str, output_dir: str = "td_lang_outputs") -> Path: """Compile a .td file to Python.""" return TDExecutor(output_dir=output_dir).compile(td_path) def run_td_file(td_path: str, output_dir: str = "td_lang_outputs", dry_run: bool = False) -> dict: """Compile and run a .td file.""" return TDExecutor(output_dir=output_dir).run(td_path, dry_run=dry_run)