"""Executes code files from the editor in isolated subprocesses. Supports Python (.py) via the system Python interpreter and LaTeX (.tex) via pdflatex. All execution is time-bounded by RUN_TIMEOUT to prevent runaway processes from blocking the UI indefinitely. """ import subprocess import sys from pathlib import Path from backend.managers.debug_logger import get_logger logger = get_logger(__name__) # Maximum time (seconds) a subprocess is allowed to run before being killed. RUN_TIMEOUT = 30 class ExecutionEngine: """Runs files from the editor in a subprocess and returns the output. Currently supports Python (.py) and LaTeX (.tex) files. Returns a dict with keys: stdout, stderr, rc (return code). """ def __init__(self): pass def run_code(self, active_file: Path) -> dict: """Execute the given file and return its output. Args: active_file: Absolute path to the file that should be run. Returns: {"stdout": str, "stderr": str, "rc": int} rc == 0 means success, anything else is an error. """ suffix = active_file.suffix current_dir = active_file.parent.resolve() # Build the shell command depending on file type if suffix == ".py": cmd = [sys.executable, active_file.name] else: return {"stdout": "", "stderr": f"Unsupported file type: {suffix}", "rc": 1} logger.info("Running file %s with suffix %s", active_file.name, suffix) try: proc = subprocess.run( cmd, cwd=current_dir, # run inside the file's own directory capture_output=True, text=True, timeout=RUN_TIMEOUT, ) logger.info("File ran successfully.") return self.capture_output(proc) except subprocess.TimeoutExpired: logger.warning("Time out afte %s s", RUN_TIMEOUT) return {"stdout": "", "stderr": f"Timed out after {RUN_TIMEOUT}s", "rc": -1} except FileNotFoundError as e: # Raised when the interpreter/compiler binary is not found on PATH logger.warning("Interpreter/compiler binary is not found on PATH: %s", e) return {"stdout": "", "stderr": str(e), "rc": -1} except Exception as e: logger.exception("Error while running %s: %s", active_file.name, e) return {"stdout": "", "stderr": str(e), "rc": -1} def capture_output(self, proc: subprocess.CompletedProcess) -> dict: """Extract stdout, stderr, and return code from a completed subprocess. Args: proc: The CompletedProcess returned by subprocess.run(). Returns: {"stdout": str, "stderr": str, "rc": int} with whitespace stripped. """ return { "stdout": proc.stdout.strip(), "stderr": proc.stderr.strip(), "rc": proc.returncode, }