Function get_assert

examples/integration-e2b/validate_and_run_code_e2b.py:190–284

Execute the generated function in E2B and compare stdout with expected output.

(output, context)

def get_assert(output, context):
    """Execute the generated function in E2B and compare stdout with expected output."""
    task_id = context.get("id", str(time.time()))
    provider = context.get("provider", "unknown")
    model = context.get("model", "unknown")

    fn_name = context["vars"]["function_name"]
    test_input = context["vars"]["test_input"]
    expected = str(context["vars"]["expected_output"])

    function_code = _extract_function(output, fn_name)
    if not function_code:
        snippet = output.strip()[:300].replace("\n", " ")
        write_metrics(
            task_id, provider, model, False, 0.0, extra={"reason": "no_code_found"}
        )
        return {
            "pass": False,
            "score": 0,
            "reason": f"No Python code block found (first 300 chars: {snippet})",
        }

    # Pre-exec safety
    if is_unsafe(function_code):
        write_metrics(
            task_id, provider, model, False, 0.0, extra={"reason": "unsafe_pattern"}
        )
        return {
            "pass": False,
            "score": 0,
            "reason": "Unsafe pattern detected in generated code",
        }

    test_program = f"""{function_code}

print({fn_name}({test_input}))
"""

    start = time.time()
    with Sandbox.create() as sbx:
        try:
            res = _run_code_in_sandbox(sbx, test_program)
        except Exception as e:
            duration = time.time() - start
            write_metrics(
                task_id, provider, model, False, duration, extra={"error": str(e)}
            )
            return {
                "pass": False,
                "score": 0,
                "reason": f"Sandbox execution error: {e}",
            }

    duration = time.time() - start
    stdout, stderr = _stdout_from_result(res)
    err_obj = getattr(res, "error", None)

    if err_obj or stderr: