code_executor.py

"""
RUBRA — code_executor.py
Safe sandboxed code execution engine.

Supports: Python, JavaScript (Node.js), Bash
Features:
- Timeout protection (30s max)
- Memory limit
- AST-based dangerous-import/call blocking (Python) — not regex
- Restricted builtins actually ENFORCED at exec() time, not just declared
- JavaScript static safety check (dangerous modules/keywords)
- stdout/stderr capture
- Auto-fix loop: if error → AI fixes → re-run (max 3 attempts)

SECURITY NOTE: regex-only checks (the old `BLOCKED_PATTERNS` approach) are
trivially bypassed — `__import__('os').system(...)`, `getattr(__builtins__,
'ex'+'ec')(...)`, string concatenation, `\x5f\x5fimport\x5f\x5f`, etc. all
sail past a substring/regex scan. AST inspection walks the actual parsed
structure, so it can't be fooled by how the dangerous call is *spelled* —
only by what it *is*. It's still not a perfect sandbox (a determined
attacker with a Python interpreter and no import restrictions can chain
allowed builtins in surprising ways), so this is defense-in-depth layered
on top of: subprocess isolation, a timeout, output caps, and — the part
that was previously missing — builtins that are actually restricted at
the point of execution, not just defined and ignored.
"""

import os
import re
import ast
import sys
import json
import time
import signal
import asyncio
import logging
import tempfile
import subprocess
from pathlib import Path
from typing import Optional

log = logging.getLogger("rubra.executor")

# ── Safety config ─────────────────────────────────────────────────────────────
TIMEOUT_SECONDS = 30
MAX_OUTPUT_CHARS = 8000
MAX_FIX_ATTEMPTS = 3

# ═══════════════════════════════════════════════════════
#  PYTHON — AST-based safety check
# ═══════════════════════════════════════════════════════

# Modules that are never importable from sandboxed code. This list mirrors
# (and supersedes) the old regex BLOCKED_PATTERNS, but is enforced by
# walking Import/ImportFrom AST nodes instead of grepping source text, so it
# also catches `import os as o`, `from os import system as sys_call`, etc.
BLOCKED_MODULES = {
    "os", "subprocess", "sys", "shutil", "socket", "ctypes",
    "multiprocessing", "threading", "pty", "pickle", "marshal", "shelve",
    "importlib", "requests", "urllib", "urllib2", "httpx", "aiohttp",
    "ftplib", "smtplib", "telnetlib", "poplib", "imaplib", "resource",
    "sysconfig", "code", "codeop", "pdb", "pip", "site", "distutils",
    "asyncio",  # sandboxed code shouldn't be spinning its own event loop
    "signal", "fcntl", "mmap", "posix", "pwd", "grp", "tty", "termios",
    "webbrowser", "platform",
}

# `sys` is allowed by the SAFE_PYTHON_PRELUDE for things like sys.exit-free
# introspection, but sys itself exposes far too much (sys.modules,
# sys._getframe, etc.) to hand to untrusted code — keep it out of both the
# prelude's real import and the blocklist exemptions.

# Builtin functions never allowed to be *called*, even though some remain
# reachable as names in the restricted builtins dict for introspection
# reasons elsewhere in the codebase.
BLOCKED_CALL_NAMES = {
    "eval", "exec", "compile", "__import__", "open", "input",
    "globals", "locals", "vars", "breakpoint", "help",
    "memoryview", "exit", "quit",
}

# Attribute/method names on ANY object that are classic sandbox-escape
# vectors (walking the object graph back to builtins/subclasses/frames)
# e.g. `().__class__.__bases__[0].__subclasses__()`.
BLOCKED_ATTR_NAMES = {
    "__globals__", "__subclasses__", "__bases__", "__mro__", "__base__",
    "__builtins__", "__import__", "__loader__", "__spec__", "__code__",
    "__closure__", "__func__", "__self__", "__dict__", "__getattribute__",
    "__class__",
}

# os.system / subprocess.run / shutil.rmtree style dotted calls — kept as a
# belt-and-suspenders check in case a blocked module somehow gets imported
# indirectly (e.g. via a permitted module re-exporting it).
BLOCKED_DOTTED_CALLS = {
    ("os", "system"), ("os", "popen"), ("os", "fork"), ("os", "kill"),
    ("subprocess", "run"), ("subprocess", "Popen"), ("subprocess", "call"),
    ("shutil", "rmtree"), ("shutil", "move"),
}


class _PySafetyVisitor(ast.NodeVisitor):
    """Walks the parsed AST once and collects every violation found,
    rather than bailing on the first one — a single clear error listing
    everything wrong is more useful than a whack-a-mole retry loop."""

    def __init__(self):
        self.violations: list[str] = []

    def visit_Import(self, node: ast.Import):
        for alias in node.names:
            root = alias.name.split(".")[0]
            if root in BLOCKED_MODULES:
                self.violations.append(f"import of blocked module '{alias.name}'")
        self.generic_visit(node)

    def visit_ImportFrom(self, node: ast.ImportFrom):
        root = (node.module or "").split(".")[0]
        if root in BLOCKED_MODULES:
            self.violations.append(f"import from blocked module '{node.module}'")
        self.generic_visit(node)

    def visit_Call(self, node: ast.Call):
        func = node.func
        if isinstance(func, ast.Name) and func.id in BLOCKED_CALL_NAMES:
            self.violations.append(f"call to blocked builtin '{func.id}()'")
        elif isinstance(func, ast.Attribute):
            attr = func.attr
            if attr in BLOCKED_ATTR_NAMES:
                self.violations.append(f"call to blocked attribute '.{attr}()'")
            if isinstance(func.value, ast.Name) and (func.value.id, attr) in BLOCKED_DOTTED_CALLS:
                self.violations.append(f"call to blocked '{func.value.id}.{attr}()'")
        self.generic_visit(node)

    def visit_Attribute(self, node: ast.Attribute):
        if node.attr in BLOCKED_ATTR_NAMES:
            self.violations.append(f"access to blocked attribute '.{node.attr}'")
        self.generic_visit(node)

    def visit_Name(self, node: ast.Name):
        if node.id == "__import__":
            self.violations.append("reference to '__import__'")
        self.generic_visit(node)


def _python_ast_safety_check(code: str) -> tuple[bool, str]:
    """Returns (is_safe, reason). A SyntaxError is treated as unsafe rather
    than silently passed through to the interpreter — malformed input has
    no business reaching exec()."""
    try:
        tree = ast.parse(code)
    except SyntaxError as e:
        return False, f"Syntax error: {e}"

    visitor = _PySafetyVisitor()
    visitor.visit(tree)
    if visitor.violations:
        # De-dupe while preserving order, cap so the error stays readable.
        seen, uniq = set(), []
        for v in visitor.violations:
            if v not in seen:
                seen.add(v)
                uniq.append(v)
        return False, "Blocked: " + "; ".join(uniq[:5])
    return True, ""


# ═══════════════════════════════════════════════════════
#  JAVASCRIPT — static safety check
# ═══════════════════════════════════════════════════════
# No JS parser dependency is available in this environment, so this is a
# careful regex pass tuned to the specific escape hatches Node.js exposes
# that Python doesn't: requiring child_process/fs/net, process.exit /
# process.kill, dynamic require via a variable, vm/Function-based eval
# equivalents, and Buffer-based shellcode-ish tricks. Like the old Python
# regex layer this is bypassable by a sufficiently motivated obfuscator —
# it's here to block the OBVIOUS 99% (subprocess spawn, filesystem access,
# arbitrary eval), not to be airtight.
_JS_BLOCKED_PATTERNS = [
    (r'require\s*\(\s*[\'"]child_process[\'"]\s*\)', "require('child_process')"),
    (r'require\s*\(\s*[\'"]fs[\'"]\s*\)', "require('fs')"),
    (r'require\s*\(\s*[\'"]fs/promises[\'"]\s*\)', "require('fs/promises')"),
    (r'require\s*\(\s*[\'"]net[\'"]\s*\)', "require('net')"),
    (r'require\s*\(\s*[\'"]dgram[\'"]\s*\)', "require('dgram')"),
    (r'require\s*\(\s*[\'"]cluster[\'"]\s*\)', "require('cluster')"),
    (r'require\s*\(\s*[\'"]vm[\'"]\s*\)', "require('vm')"),
    (r'require\s*\(\s*[\'"]repl[\'"]\s*\)', "require('repl')"),
    (r'require\s*\(\s*[\'"]worker_threads[\'"]\s*\)', "require('worker_threads')"),
    (r'\bimport\s*\(\s*[\'"]child_process[\'"]', "dynamic import('child_process')"),
    (r'\bimport\s*\(\s*[\'"]fs[\'"]', "dynamic import('fs')"),
    (r'\bprocess\s*\.\s*exit\s*\(', "process.exit()"),
    (r'\bprocess\s*\.\s*kill\s*\(', "process.kill()"),
    (r'\bprocess\s*\.\s*binding\s*\(', "process.binding() (internal API escape hatch)"),
    (r'\bprocess\s*\.\s*mainModule\b', "process.mainModule"),
    (r'\bglobal\s*\.\s*process\s*\.\s*mainModule\b', "global.process.mainModule"),
    (r'\brequire\s*\.\s*cache\b', "require.cache tampering"),
    (r'\brequire\s*\.\s*resolve\s*\(', "require.resolve() (fs probing)"),
    (r'new\s+Function\s*\(', "new Function(...) (eval equivalent)"),
    (r'\beval\s*\(', "eval()"),
    (r'\bexecSync\b|\bspawnSync\b|\bexecFileSync\b', "sync child-process spawn"),
    (r'\b__dirname\b|\b__filename\b', "filesystem path introspection"),
]

_JS_BLOCKED_IDENTIFIERS = {"child_process", "worker_threads"}


def _js_safety_check(code: str) -> tuple[bool, str]:
    hits = []
    for pattern, label in _JS_BLOCKED_PATTERNS:
        if re.search(pattern, code):
            hits.append(label)
    for ident in _JS_BLOCKED_IDENTIFIERS:
        if re.search(r'\b' + re.escape(ident) + r'\b', code):
            hits.append(f"reference to '{ident}'")
    if hits:
        seen, uniq = set(), []
        for h in hits:
            if h not in seen:
                seen.add(h)
                uniq.append(h)
        return False, "Blocked: " + "; ".join(uniq[:5])
    return True, ""


# ═══════════════════════════════════════════════════════
#  RESTRICTED BUILTINS — actually enforced
# ═══════════════════════════════════════════════════════
# Names intentionally present so ordinary code still works.
_ALLOWED_BUILTIN_NAMES = {
    "abs", "all", "any", "ascii", "bin", "bool", "bytearray", "bytes",
    "callable", "chr", "classmethod", "complex", "delattr", "dict", "dir",
    "divmod", "enumerate", "filter", "float", "format", "frozenset",
    "getattr", "hasattr", "hash", "hex", "id", "int", "isinstance",
    "issubclass", "iter", "len", "list", "map", "max", "min", "next",
    "object", "oct", "ord", "pow", "print", "property", "range", "repr",
    "reversed", "round", "set", "setattr", "slice", "sorted",
    "staticmethod", "str", "sum", "super", "tuple", "type", "zip",
    "True", "False", "None", "NotImplemented", "Ellipsis",
    # Common, non-dangerous exception types code legitimately catches/raises
    "Exception", "BaseException", "ValueError", "TypeError", "KeyError",
    "IndexError", "AttributeError", "StopIteration", "ZeroDivisionError",
    "RuntimeError", "NotImplementedError", "ArithmeticError",
    "OverflowError", "FloatingPointError", "AssertionError",
    "LookupError", "NameError", "UnboundLocalError", "RecursionError",
    "GeneratorExit", "KeyboardInterrupt", "StopAsyncIteration",
}

SAFE_PYTHON_PRELUDE = """
import sys as _sys
import math
import json
import re
import random
import datetime
import itertools
import functools
import collections
import statistics
from typing import *

# ── Restricted builtins — this is the part that must actually be ENFORCED,
# not just declared. Previously `_safe_builtins` was built here but never
# wired into how the user code actually ran, so it was pure decoration.
# The real enforcement now happens where this prelude is exec()'d: the
# runner wraps USER CODE in exec(compile(...), {"__builtins__": _safe_builtins})
# so the restricted mapping is what Python's name resolution actually sees.
import builtins as _builtins_module
_ALLOWED = %r
_safe_builtins = {k: v for k, v in vars(_builtins_module).items() if k in _ALLOWED}

# BUG FIX (found via testing): a __builtins__ dict with NO '__import__' key
# doesn't just block *dangerous* imports — it breaks the `import` statement
# entirely, since Python's IMPORT_NAME opcode looks up __import__ on
# __builtins__ unconditionally. Without this, even "import math" inside the
# restricted exec() raised "ImportError: __import__ not found", making the
# sandbox unable to run almost any real code. Fix: install a WHITELIST-
# CHECKED __import__ — a second, independent enforcement layer alongside
# the AST blacklist check in _python_ast_safety_check(). The two layers
# check different things (static text vs. runtime call) and use different
# list philosophies (blacklist vs. whitelist) on purpose — a gap or bug in
# either one alone still leaves the other standing.
_ALLOWED_RUNTIME_IMPORTS = {
    "math", "json", "re", "random", "datetime", "itertools", "functools",
    "collections", "statistics", "string", "time", "decimal", "fractions",
    "heapq", "bisect", "array", "copy", "typing", "dataclasses", "enum",
    "textwrap", "unicodedata", "difflib", "csv", "io", "operator", "uuid",
    "hashlib", "base64", "struct", "queue", "abc", "numbers", "calendar",
}
def _safe_import(name, globals=None, locals=None, fromlist=(), level=0):
    root = name.split(".")[0]
    if root not in _ALLOWED_RUNTIME_IMPORTS:
        raise ImportError(f"Import of '{name}' is not allowed in the sandbox")
    return __import__(name, globals, locals, fromlist, level)
_safe_builtins["__import__"] = _safe_import
del _sys, _builtins_module, _ALLOWED
""" % (_ALLOWED_BUILTIN_NAMES,)


# ── Language detection ────────────────────────────────────────────────────────
def detect_language(code: str, hint: str = "") -> str:
    """Detect programming language from code or hint."""
    hint = hint.lower()
    if hint in ("python", "py"):   return "python"
    if hint in ("javascript", "js", "node"): return "javascript"
    if hint in ("bash", "sh", "shell"): return "bash"

    # Auto-detect from code
    if re.search(r'^\s*(def |import |from .* import|print\(|if __name__)', code, re.M):
        return "python"
    if re.search(r'(const |let |var |=>|console\.log|require\(|async function)', code):
        return "javascript"
    if re.search(r'^#!(\/usr\/bin\/(env )?bash|\/bin\/sh)', code):
        return "bash"
    if re.search(r'function\s+\w+\s*\(', code):
        return "javascript"

    return "python"  # default


# ── Safety check ─────────────────────────────────────────────────────────────
def safety_check(code: str, lang: str) -> tuple[bool, str]:
    """Returns (is_safe, reason)"""
    if lang == "python":
        return _python_ast_safety_check(code)

    if lang == "javascript":
        return _js_safety_check(code)

    # Block rm -rf style commands
    if lang == "bash":
        dangerous = [r'rm\s+-rf', r'mkfs', r'dd\s+if=', r':\(\)\{', r'fork bomb']
        for p in dangerous:
            if re.search(p, code, re.IGNORECASE):
                return False, f"Dangerous bash command blocked"

    return True, ""


# ── Python executor ───────────────────────────────────────────────────────────
async def run_python(code: str) -> dict:
    """
    Run Python code in a subprocess with timeout.

    The subprocess itself is the outer sandbox boundary (process isolation,
    kill on timeout, capped output). Inside that subprocess, the user code
    is executed via exec() against a globals dict whose "__builtins__" is
    the RESTRICTED mapping built by SAFE_PYTHON_PRELUDE — this is what
    actually enforces the allowlist, rather than just defining it and
    letting the script run with full normal builtins regardless.
    """
    # User code is embedded as a JSON string literal (not f-string-spliced
    # raw text) so there's no risk of the user's code breaking out of the
    # wrapper via a stray triple-quote or backslash sequence.
    user_code_literal = json.dumps(code)

    runner = (
        SAFE_PYTHON_PRELUDE
        + "\n# === RUNNER ===\n"
        + "_user_code = " + user_code_literal + "\n"
        + "_restricted_globals = {\n"
        + "    '__builtins__': _safe_builtins, '__name__': '__main__',\n"
        + "    'math': math, 'json': json, 're': re, 'random': random,\n"
        + "    'datetime': datetime, 'itertools': itertools,\n"
        + "    'functools': functools, 'collections': collections,\n"
        + "    'statistics': statistics,\n"
        + "}\n"
        + "exec(compile(_user_code, '<user_code>', 'exec'), _restricted_globals)\n"
    )

    with tempfile.NamedTemporaryFile(mode='w', suffix='.py',
                                     delete=False, dir='/tmp') as f:
        f.write(runner)
        tmp_path = f.name

    try:
        proc = await asyncio.create_subprocess_exec(
            sys.executable, tmp_path,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            cwd='/tmp',
        )
        try:
            stdout, stderr = await asyncio.wait_for(
                proc.communicate(), timeout=TIMEOUT_SECONDS
            )
        except asyncio.TimeoutError:
            proc.kill()
            return {
                "success": False,
                "stdout":  "",
                "stderr":  f"⏱ Execution timed out after {TIMEOUT_SECONDS}s",
                "exit_code": -1,
                "lang": "python",
            }

        out = stdout.decode("utf-8", errors="replace")[:MAX_OUTPUT_CHARS]
        err = stderr.decode("utf-8", errors="replace")[:2000]

        # Filter out runner-wrapper noise from stderr tracebacks so users see
        # their own code's error, not our wrapper's internal frame.
        err_lines = [l for l in err.splitlines()
                     if 'SAFE_PYTHON_PRELUDE' not in l
                     and 'tmp_path' not in l
                     and '_restricted_globals' not in l
                     and '/tmp/' not in l.lower()[:20]]
        err = "\n".join(err_lines)

        return {
            "success":   proc.returncode == 0,
            "stdout":    out,
            "stderr":    err,
            "exit_code": proc.returncode,
            "lang":      "python",
        }
    finally:
        try: os.unlink(tmp_path)
        except: pass


# ── JavaScript executor ───────────────────────────────────────────────────────
async def run_javascript(code: str) -> dict:
    """Run JavaScript with Node.js."""
    # Check node available
    try:
        check = await asyncio.create_subprocess_exec(
            "node", "--version",
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        await check.communicate()
    except FileNotFoundError:
        return {"success": False, "stdout": "",
                "stderr": "Node.js not installed on this server",
                "exit_code": -1, "lang": "javascript"}

    with tempfile.NamedTemporaryFile(mode='w', suffix='.js',
                                     delete=False, dir='/tmp') as f:
        f.write(code)
        tmp_path = f.name

    try:
        proc = await asyncio.create_subprocess_exec(
            "node", "--max-old-space-size=128", tmp_path,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            cwd='/tmp',
        )
        try:
            stdout, stderr = await asyncio.wait_for(
                proc.communicate(), timeout=TIMEOUT_SECONDS
            )
        except asyncio.TimeoutError:
            proc.kill()
            return {"success": False, "stdout": "",
                    "stderr": f"⏱ Timeout after {TIMEOUT_SECONDS}s",
                    "exit_code": -1, "lang": "javascript"}

        return {
            "success":   proc.returncode == 0,
            "stdout":    stdout.decode("utf-8", errors="replace")[:MAX_OUTPUT_CHARS],
            "stderr":    stderr.decode("utf-8", errors="replace")[:2000],
            "exit_code": proc.returncode,
            "lang":      "javascript",
        }
    finally:
        try: os.unlink(tmp_path)
        except: pass


# ── Bash executor ─────────────────────────────────────────────────────────────
async def run_bash(code: str) -> dict:
    """Run bash in restricted environment."""
    with tempfile.NamedTemporaryFile(mode='w', suffix='.sh',
                                     delete=False, dir='/tmp') as f:
        f.write("#!/bin/bash\nset -euo pipefail\n" + code)
        tmp_path = f.name
    os.chmod(tmp_path, 0o700)

    try:
        proc = await asyncio.create_subprocess_exec(
            "bash", tmp_path,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            cwd='/tmp',
        )
        try:
            stdout, stderr = await asyncio.wait_for(
                proc.communicate(), timeout=15  # shorter for bash
            )
        except asyncio.TimeoutError:
            proc.kill()
            return {"success": False, "stdout": "",
                    "stderr": "⏱ Bash timeout after 15s",
                    "exit_code": -1, "lang": "bash"}

        return {
            "success":   proc.returncode == 0,
            "stdout":    stdout.decode("utf-8", errors="replace")[:MAX_OUTPUT_CHARS],
            "stderr":    stderr.decode("utf-8", errors="replace")[:2000],
            "exit_code": proc.returncode,
            "lang":      "bash",
        }
    finally:
        try: os.unlink(tmp_path)
        except: pass


# ── Main execute function ─────────────────────────────────────────────────────
async def execute_code(code: str, lang: str = "") -> dict:
    """
    Execute code safely. Returns execution result dict.
    """
    lang = detect_language(code, lang)

    # Safety check
    safe, reason = safety_check(code, lang)
    if not safe:
        return {
            "success":   False,
            "stdout":    "",
            "stderr":    f"🚫 Blocked: {reason}",
            "exit_code": -1,
            "lang":      lang,
            "blocked":   True,
        }

    start = time.time()
    if lang == "python":
        result = await run_python(code)
    elif lang == "javascript":
        result = await run_javascript(code)
    elif lang == "bash":
        result = await run_bash(code)
    else:
        result = {"success": False, "stdout": "",
                  "stderr": f"Unsupported language: {lang}",
                  "exit_code": -1, "lang": lang}

    result["elapsed_ms"] = int((time.time() - start) * 1000)
    return result


# ── Auto-fix loop (Kimi-style) ────────────────────────────────────────────────
async def execute_with_autofix(
    code: str,
    lang: str = "",
    task_description: str = "",
    llm_func = None,
) -> dict:
    """
    Execute code. If error → ask AI to fix → re-execute.
    Repeats up to MAX_FIX_ATTEMPTS times.
    Returns final result with fix history.
    """
    lang     = detect_language(code, lang)
    attempts = []
    current_code = code

    for attempt in range(1, MAX_FIX_ATTEMPTS + 1):
        log.info(f"[EXECUTOR] Attempt {attempt}/{MAX_FIX_ATTEMPTS} — {lang}")
        result = await execute_code(current_code, lang)
        result["attempt"] = attempt
        result["code"]    = current_code
        attempts.append(result)

        if result["success"]:
            log.info(f"[EXECUTOR] ✅ Success on attempt {attempt}")
            break

        if result.get("blocked"):
            break

        # Error — ask AI to fix (if llm_func provided)
        if llm_func and attempt < MAX_FIX_ATTEMPTS:
            log.info(f"[EXECUTOR] ⚡ Auto-fixing error...")
            error_msg = result.get("stderr", "") or result.get("stdout", "")

            fix_prompt = f"""Fix this {lang} code that has an error.

ORIGINAL TASK: {task_description or 'Execute the code correctly'}

CODE:
```{lang}
{current_code}
```

ERROR:
```
{error_msg[:500]}
```

Return ONLY the fixed code. No explanation. No markdown fences."""

            msgs = [
                {"role": "system", "content": f"You are an expert {lang} debugger. Fix the error and return only the corrected code."},
                {"role": "user",   "content": fix_prompt},
            ]

            fixed_code = ""
            try:
                async for tok in llm_func(msgs, mode="coding"):
                    fixed_code += tok
                # Strip markdown fences if AI adds them
                fixed_code = re.sub(r'^```\w*\n?', '', fixed_code.strip())
                fixed_code = re.sub(r'\n?```$', '', fixed_code.strip())
                if fixed_code:
                    current_code = fixed_code
                    log.info(f"[EXECUTOR] AI fix applied, retrying...")
                else:
                    break
            except Exception as e:
                log.warning(f"[EXECUTOR] AI fix failed: {e}")
                break

    final = attempts[-1].copy()
    final["attempts"]     = len(attempts)
    final["fix_history"]  = attempts if len(attempts) > 1 else []
    final["auto_fixed"]   = len(attempts) > 1 and attempts[-1]["success"]
    return final


# ── Format result for SSE ─────────────────────────────────────────────────────
def format_execution_result(result: dict) -> str:
    """Format execution result as readable text for frontend."""
    lang      = result.get("lang", "")
    success   = result.get("success", False)
    stdout    = result.get("stdout", "").strip()
    stderr    = result.get("stderr", "").strip()
    elapsed   = result.get("elapsed_ms", 0)
    attempts  = result.get("attempts", 1)
    auto_fix  = result.get("auto_fixed", False)

    lines = []

    if auto_fix:
        lines.append(f"🔧 Auto-fixed after {attempts} attempt(s)\n")

    if success:
        lines.append(f"✅ **Executed** ({lang}, {elapsed}ms)")
        if stdout:
            lines.append(f"\n**Output:**\n```\n{stdout}\n```")
        else:
            lines.append("\n*(No output)*")
    else:
        if result.get("blocked"):
            lines.append(f"🚫 **Blocked:** {stderr}")
        else:
            lines.append(f"❌ **Error** ({lang}, exit {result.get('exit_code', '?')})")
            if stderr:
                lines.append(f"\n```\n{stderr[:800]}\n```")
            if stdout:
                lines.append(f"\n**Partial output:**\n```\n{stdout[:400]}\n```")

    return "\n".join(lines)