"""Runs a Custom Node's script and explains what went wrong. The run itself is the same one a task run does (app/crud/tasks.py's custom_code node): a separate OS process with a hard timeout and a stripped environment — NOT a sandbox — fed the node's resolved inputs as JSON on stdin, whose last printed line (when it is JSON) is the node's result. `diagnose` turns a run (or a syntax check) into problems a script author can act on: which line, what the error means in plain words, and how to fix it — used by the Task Designer's code editor (POST /tasks/code/run). """ import json import os import re import shutil import subprocess import sys import tempfile import time from dataclasses import dataclass, field from pathlib import Path from typing import Any from app.crud.params import apply_output_params TIMEOUT_SECONDS = 15 @dataclass class CodeRun: ok: bool detail: str result: Any = None stdout: str = "" stderr: str = "" exit_code: int | None = None timed_out: bool = False duration_ms: int = 0 result_is_json: bool = False @dataclass class Problem: severity: str # "error" | "warning" | "info" message: str # what happened, in plain words fix: str # what to change line: int | None = None column: int | None = None @dataclass class Diagnosis: problems: list[Problem] = field(default_factory=list) def add(self, severity: str, message: str, fix: str, line: int | None = None, column: int | None = None) -> None: self.problems.append(Problem(severity, message, fix, line, column)) def run_code(language: str, code: str, inputs: dict) -> CodeRun: with tempfile.TemporaryDirectory() as tmp: tmp_path = Path(tmp) if language == "node": node_bin = shutil.which("node") if not node_bin: return CodeRun(False, "node is not available on this server") script = tmp_path / "script.js" cmd = [node_bin, str(script)] else: script = tmp_path / "script.py" cmd = [sys.executable, str(script)] script.write_text(code, encoding="utf-8") env = {"PATH": os.environ.get("PATH", ""), "PYTHONIOENCODING": "utf-8"} # Windows processes need to know where the OS lives — without # SYSTEMROOT, Node can't even start (its crypto init fails) and some # Python modules break. Only these locations are passed, no secrets. for name in ("SYSTEMROOT", "WINDIR"): if os.environ.get(name): env[name] = os.environ[name] started = time.monotonic() try: proc = subprocess.run( cmd, input=json.dumps(inputs, default=str), capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=TIMEOUT_SECONDS, cwd=tmp_path, env=env, ) except subprocess.TimeoutExpired as exc: return CodeRun( False, f"timed out after {TIMEOUT_SECONDS}s", stdout=_text(exc.stdout), stderr=_text(exc.stderr), timed_out=True, duration_ms=int((time.monotonic() - started) * 1000), ) duration = int((time.monotonic() - started) * 1000) # Temp paths mean nothing to the author — show just the script's name. stderr = proc.stderr.replace(str(script), script.name).replace(str(tmp_path), "") if proc.returncode != 0: return CodeRun(False, f"exited {proc.returncode}: {stderr[-500:]}", stdout=proc.stdout, stderr=stderr, exit_code=proc.returncode, duration_ms=duration) last_line = next((line for line in reversed(proc.stdout.splitlines()) if line.strip()), "") try: result, is_json = json.loads(last_line), True except (json.JSONDecodeError, ValueError): result, is_json = proc.stdout[-2000:], False return CodeRun(True, "script exited 0", result, proc.stdout, stderr, 0, duration_ms=duration, result_is_json=is_json) def _text(value: Any) -> str: if value is None: return "" return value.decode("utf-8", "replace") if isinstance(value, bytes) else str(value) def logs_of(run: CodeRun) -> list[str]: """Everything the script printed except the line used as its result.""" lines = run.stdout.splitlines() if run.ok and run.result_is_json: for i in range(len(lines) - 1, -1, -1): if lines[i].strip(): return lines[:i] return lines # ---- syntax --------------------------------------------------------------------------------------- def check_python_syntax(code: str, diagnosis: Diagnosis) -> bool: try: compile(code, "script.py", "exec") return True except SyntaxError as exc: # includes IndentationError / TabError line, col, msg = exc.lineno, exc.offset, exc.msg or "invalid syntax" diagnosis.add("error", f"Syntax error on line {line}: {msg}", _python_syntax_fix(exc, msg, line), line, col) return False def _python_syntax_fix(exc: SyntaxError, msg: str, line: int | None) -> str: where = f"line {line}" if line else "the marked line" if isinstance(exc, TabError) or "inconsistent use of tabs" in msg: return f"Indentation mixes tabs and spaces around {where}. Re-indent that block using spaces only (4 per level)." if "expected an indented block" in msg: return f"The line before {where} ends with ':' (if / for / def / with …), so {where} must be indented one level (4 spaces) more than it." if "unexpected indent" in msg: return f"{where.capitalize()} is indented more than the line above it, but nothing opened a block. Line it up with the surrounding code." if "unindent does not match" in msg: return f"{where.capitalize()} is dedented to a level no earlier line used. Make its indentation match the block it belongs to (multiples of 4 spaces)." if "unterminated string" in msg or "EOL while scanning" in msg or "unterminated triple-quoted" in msg: return f"A quote opened on {where} is never closed. Add the matching \" or ' at the end of the text." if "was never closed" in msg: return f"A bracket opened on {where} is never closed. Add the matching ) ] or }} (often at the end of that statement)." if "unmatched" in msg or "does not match opening" in msg: return f"There is a closing bracket on {where} with no matching opening one. Remove it or add the opening bracket." if "expected ':'" in msg: return f"Add ':' at the end of {where} — if / elif / else / for / while / def / class / try lines must end with a colon." if "Perhaps you forgot a comma" in msg: return f"Two values sit side by side on {where}. Separate them with a comma." if "invalid syntax" in msg and line and line > 1: return f"Check {where} and the line just before it for a missing ':', comma, bracket or quote, or a misspelled keyword." return f"Fix the code on {where} so it is valid Python." # ---- runtime errors ------------------------------------------------------------------------------- def _python_error_line(stderr: str) -> int | None: frames = re.findall(r'File "script\.py", line (\d+)', stderr) return int(frames[-1]) if frames else None def _last_error_line(stderr: str) -> str: return next((line.strip() for line in reversed(stderr.splitlines()) if line.strip()), "") def _diagnose_python_failure(run: CodeRun, inputs: dict, diagnosis: Diagnosis) -> None: line = _python_error_line(run.stderr) error = _last_error_line(run.stderr) kind, _, text = error.partition(":") kind, text = kind.strip().split(".")[-1], text.strip() at = f" (line {line})" if line else "" keys = ", ".join(sorted(inputs)) or "none" if kind == "KeyError": name = text.strip("'\"") if name not in inputs: fix = ( f"This run's input has no '{name}' (it has: {keys}). Declare an input named '{name}' on this node, " f"or read it with context.get(\"{name}\") so a missing value is allowed." ) else: fix = f"'{name}' isn't a key of the value being read{at}. Print that value to see its keys, or use .get(\"{name}\")." diagnosis.add("error", f"KeyError{at}: the key '{name}' doesn't exist.", fix, line) elif kind == "NameError": name = re.search(r"name '([^']+)'", text) name = name.group(1) if name else "the name" fix = ( f"'{name}' is a node input — inputs aren't variables of their own; read it from the loaded input, e.g. context[\"{name}\"]." if name in inputs else f"'{name}' is used before it is defined. Check the spelling, assign it first, or add the missing import." ) diagnosis.add("error", f"NameError{at}: {text}", fix, line) elif kind in ("ModuleNotFoundError", "ImportError"): module = re.search(r"'([^']+)'", text) module = module.group(1) if module else "that module" diagnosis.add( "error", f"{kind}{at}: {text}", f"'{module}' isn't installed on this server. Use the Python standard library instead, or ask an administrator to install it into the backend's Python environment.", line, ) elif kind == "JSONDecodeError": diagnosis.add( "error", f"JSONDecodeError{at}: {text}", "Reading JSON failed. Read the node's input exactly once with context = json.load(sys.stdin) — a second read of sys.stdin is empty. " "If you are parsing an input value, print it first to check it is JSON.", line, ) elif kind == "TypeError": if "concatenate" in text or "unsupported operand" in text or "can't multiply" in text: fix = "The values are of different types (e.g. text and a number). Convert first — int(x), float(x) or str(x) — or declare the input with the right type." elif "not subscriptable" in text: fix = "You are indexing something that isn't a list or dict (often None). Check that the value exists before using [ ]." elif "not JSON serializable" in text: fix = "json.dumps can't write that value. Convert it first, e.g. str(value) for dates, or pass default=str to json.dumps." else: fix = f"Check the types of the values used on line {line}." if line else "Check the types of the values used." diagnosis.add("error", f"TypeError{at}: {text}", fix, line) elif kind == "AttributeError" and "NoneType" in text: diagnosis.add( "error", f"AttributeError{at}: {text}", "A value on that line is None — usually a missing input or a lookup that found nothing. Check it (if value is not None:) before using it.", line, ) elif kind == "ValueError" and "invalid literal" in text: diagnosis.add("error", f"ValueError{at}: {text}", "A text value isn't a number. Check the input's value, or guard the conversion with try/except.", line) elif kind == "ZeroDivisionError": diagnosis.add("error", f"ZeroDivisionError{at}: {text}", "A divisor is 0. Check it before dividing.", line) elif kind == "IndexError": diagnosis.add("error", f"IndexError{at}: {text}", "The list is shorter than the position used. Check len(...) first, or loop over the list instead.", line) elif kind == "SystemExit" or (run.exit_code not in (None, 0) and not error): diagnosis.add("error", f"The script stopped with exit code {run.exit_code}.", "Remove the sys.exit(...) call (or exit with 0) and print the result as JSON instead.", line) else: diagnosis.add("error", f"{error or 'The script failed'}{at}", f"Fix the code on line {line}." if line else "See the error output below.", line) def _node_error_location(stderr: str) -> tuple[int | None, int | None]: m = re.search(r"script\.js:(\d+):(\d+)", stderr) if m: return int(m.group(1)), int(m.group(2)) m = re.search(r"script\.js:(\d+)", stderr) return (int(m.group(1)), None) if m else (None, None) def _diagnose_node_failure(run: CodeRun, inputs: dict, diagnosis: Diagnosis) -> None: line, col = _node_error_location(run.stderr) error = next((ln.strip() for ln in run.stderr.splitlines() if re.match(r"\s*\w*Error\b", ln)), _last_error_line(run.stderr)) kind, _, text = error.partition(":") kind, text = kind.strip(), text.strip() at = f" (line {line})" if line else "" if kind == "SyntaxError": if "JSON" in text: fix = ( "Parsing JSON failed. Parse the node's input only after it has been read completely — inside " 'process.stdin.on("end", () => { const ctx = JSON.parse(input); … }).' ) elif "Unexpected end of input" in text: fix = "A bracket, brace or string opened earlier is never closed. Add the missing ) } ] or quote — often at the end of the script." elif "missing ) after argument list" in text: fix = f"Close the function call on line {line} with ')' — check for a missing comma or quote in its arguments too." elif "Unexpected token" in text or "Invalid or unexpected token" in text: fix = f"Line {line} has a character in the wrong place — look for a missing comma, bracket or quote just before it." if line else "Look for a missing comma, bracket or quote." elif "Cannot use import statement" in text: fix = 'This runs as a CommonJS script. Use const x = require("x") instead of import.' else: fix = f"Fix the code on line {line} so it is valid JavaScript." if line else "Fix the code so it is valid JavaScript." diagnosis.add("error", f"Syntax error{at}: {text}", fix, line, col) elif kind == "ReferenceError": name = re.match(r"(\S+) is not defined", text) name = name.group(1) if name else "the name" fix = ( f"'{name}' is a node input — read it from the parsed input, e.g. ctx.{name}." if name in inputs else f"'{name}' is used before it is declared. Check the spelling, declare it with const/let, or require() it." ) diagnosis.add("error", f"ReferenceError{at}: {text}", fix, line, col) elif kind == "TypeError" and ("of undefined" in text or "of null" in text): prop = re.search(r"reading '([^']+)'", text) what = f"before .{prop.group(1)} " if prop else "" diagnosis.add( "error", f"TypeError{at}: {text}", f"The value {what}is undefined/null — usually an input that isn't in this run or a missing field. " f"Check it first (if (value) …) or use optional chaining (value?.field). This run's inputs: {', '.join(sorted(inputs)) or 'none'}.", line, col, ) elif "Cannot find module" in text or "Cannot find module" in error: module = re.search(r"Cannot find module '([^']+)'", run.stderr) diagnosis.add( "error", f"Cannot find module{at}: {module.group(1) if module else ''}", "Only Node's built-in modules (fs, path, crypto, …) are available here. Remove the require, or ask an administrator to install the package on the server.", line, ) else: diagnosis.add("error", f"{error or 'The script failed'}{at}", f"Fix the code on line {line}." if line else "See the error output below.", line, col) # ---- the whole picture ---------------------------------------------------------------------------- def _reads_input(language: str, code: str) -> bool: return ("stdin" in code) if language != "node" else ("process.stdin" in code) def diagnose(language: str, code: str, run: CodeRun | None, inputs: dict, output_params: list[dict], diagnosis: Diagnosis) -> dict[str, Any] | None: """Adds the run's problems to `diagnosis`; returns the outputs the node would write (None when it wouldn't get that far).""" if inputs and not _reads_input(language, code): how = "context = json.load(sys.stdin)" if language != "node" else 'process.stdin.on("data"/"end") and JSON.parse(input)' diagnosis.add("info", "This script never reads the node's inputs.", f"Inputs arrive as one JSON object on standard input — read them with {how}.") if run is None: return None if run.timed_out: diagnosis.add( "error", f"The script ran longer than {TIMEOUT_SECONDS} seconds and was stopped.", "Look for a loop that never ends or a slow network call. " + ("A Node script must also finish: parse input in the stdin 'end' handler and don't leave timers or connections open." if language == "node" else ""), ) return None if not run.ok and run.exit_code is None: diagnosis.add("error", run.detail, "Switch the language to Python, or ask an administrator to install Node.js on the server.") return None if not run.ok: (_diagnose_node_failure if language == "node" else _diagnose_python_failure)(run, inputs, diagnosis) return None printer = "print(json.dumps({...}))" if language != "node" else "console.log(JSON.stringify({...}))" if not run.stdout.strip(): diagnosis.add("warning", "The script printed nothing, so this node has no result.", f"Print the result as JSON on the last line: {printer}.") elif not run.result_is_json: diagnosis.add( "warning", "The last printed line isn't JSON, so the node's result is the raw text.", f"For named outputs, make the very last line {printer}. Earlier prints are fine — they show as logs.", ) if not output_params: return {} ok, detail, outputs = apply_output_params(output_params, run.result) if not ok: have = ", ".join(sorted(run.result)) if isinstance(run.result, dict) else type(run.result).__name__ diagnosis.add( "error", f"Outputs: {detail}.", f"The result has: {have}. Return a key for each declared output (or set the output's Source to the right path), and match its declared type.", ) return None return outputs