agent/nodes/execute_solution.py

"""
Execution node — runs the generated solution against both test suites in a sandbox.

ROLE IN THE GRAPH
-----------------
execute_solution is the "judge" of the repair loop. It runs both test suites
in a subprocess sandbox and sets last_execution_passed. The graph routing
function (_route_after_execution) reads this flag to decide whether to proceed
to critic_review (pass) or diagnose_failure (fail).

DUAL-ORACLE TESTING STRATEGY (Fix 5)
-------------------------------------
Two independent test suites guard against different failure modes:

  1. spec_test_code (spec-blind oracle):
     - Generated by generate_spec_tests from the task description BEFORE any
       code exists. The generator never sees these tests.
     - Tests SPECIFICATION correctness: does the function behave as described?
     - Catches cases where the generator and QA both misunderstood the spec
       (adversarial tests would pass but the function is still wrong).
     - NOT regenerated on repair iterations — stays constant.

  2. current_test_code (adversarial):
     - Generated by create_adversarial_tests AFTER seeing the code.
     - Tests IMPLEMENTATION weaknesses: empty inputs, boundaries, edge cases.
     - Regenerated every iteration because the code changes.
     - Catches cases where the implementation is fragile on specific inputs.

BOTH suites must pass for last_execution_passed=True. Failure in either
generates a labeled failure summary so the debugger knows which oracle caught it.

SANDBOX EXECUTION
-----------------
Code is never exec'd in-process. sandbox.python_executor.execute() writes the
solution + tests to a temp file and runs it in a fresh subprocess with rlimit
resource limits (memory, CPU time). This isolation ensures:
  - Infinite loops don't hang the agent indefinitely.
  - Malicious or buggy code can't corrupt the agent process's state.
  - Each execution gets a clean Python environment.
"""

import logging
from typing import Any

from agent.state import AgentState, IterationRecord
from agent.events import step_event, failure_event, success_event
from sandbox.python_executor import execute, format_failure_summary

logger = logging.getLogger(__name__)


async def execute_solution(state: AgentState) -> dict[str, Any]:
    """LangGraph node: run both test suites in the sandbox and record the outcome.

    This is the only node that executes code. It does not call the LLM.

    Args:
        state: Current AgentState. Reads: iteration, current_code,
               spec_test_code, current_test_code. Also reads root_cause,
               failure_category, repair_strategy to populate the iteration record.

    Returns:
        Partial state: last_execution_passed, last_failure_summary,
                       iteration_history (appended), status, events.
    """
    iteration = state.get("iteration", 0)
    events = list(state.get("events", []))
    current_code = state["current_code"]

    spec_test_code = state.get("spec_test_code", "")
    adversarial_test_code = state.get("current_test_code", "")

    # ── Phase 1: Run spec-blind oracle tests ─────────────────────────────────
    # These tests were generated before the code existed. If the code fails
    # them, it means the implementation doesn't meet the specification — not
    # just an edge case weakness.
    spec_passed = True
    spec_summary = ""
    if spec_test_code.strip():
        events.append(step_event(
            "Running spec-blind oracle tests...",
            iteration=iteration,
        ).to_dict())
        spec_result = await execute(
            solution_code=current_code,
            test_code=spec_test_code,
        )
        spec_passed = spec_result.passed
        if not spec_passed:
            # Label the failure clearly so the debugger knows it's a spec
            # violation (deeper logical issue) not just an edge case.
            spec_summary = "SPEC TEST FAILURES:\n" + format_failure_summary(spec_result)
        logger.info(
            "Spec tests: passed=%s elapsed=%.2fs (iteration=%d)",
            spec_result.passed, spec_result.elapsed_seconds, iteration,
        )

    # ── Phase 2: Run adversarial tests ───────────────────────────────────────
    # These tests were generated after seeing the code — they target its
    # specific implementation weaknesses.
    events.append(step_event(
        "Running adversarial tests...",
        iteration=iteration,
    ).to_dict())
    adv_result = await execute(
        solution_code=current_code,
        test_code=adversarial_test_code,
    )
    adv_passed = adv_result.passed
    adv_summary = ""
    if not adv_passed:
        # Label separately from spec failures — the debugger uses this label
        # to decide whether the issue is a spec interpretation problem or an
        # implementation edge case.
        adv_summary = "ADVERSARIAL TEST FAILURES:\n" + format_failure_summary(adv_result)
    logger.info(
        "Adversarial tests: passed=%s elapsed=%.2fs (iteration=%d)",
        adv_result.passed, adv_result.elapsed_seconds, iteration,
    )

    # ── Phase 3: Combine outcomes ─────────────────────────────────────────────
    # Both must pass. If EITHER fails, the iteration failed.
    overall_passed = spec_passed and adv_passed

    # Combine failure summaries — include both if both failed, so the debugger
    # sees the full picture. Clear if passed (no failures to report).
    failure_parts = [p for p in [spec_summary, adv_summary] if p]
    combined_failure_summary = "\n\n".join(failure_parts) if failure_parts else ""
    if overall_passed:
        combined_failure_summary = ""

    # ── Phase 4: Emit outcome event ───────────────────────────────────────────
    if overall_passed:
        # SUCCESS event: Gradio UI shows a green checkmark + final code snapshot.
        # Note: status="success" here is tentative — the critic can still reject.
        events.append(success_event(code=current_code, iteration=iteration).to_dict())
        new_status = "success"
    else:
        # FAILURE event: Gradio timeline shows which tests failed and the traceback.
        events.append(failure_event(
            summary=combined_failure_summary,
            iteration=iteration,
            failed_assertions=(
                adv_result.failed_assertions if not adv_passed else []
            ),
        ).to_dict())
        new_status = "running"  # still in-flight, repair loop will handle it

    # ── Phase 5: Append to iteration_history ─────────────────────────────────
    # The debugger reads iteration_history to see how the code has changed
    # across iterations and whether repairs are making progress. We capture
    # the diagnosis fields from the PREVIOUS iteration's diagnosis (if any)
    # so the record shows what was diagnosed AND whether the repair worked.
    iteration_record: IterationRecord = {
        "iteration": iteration,
        "code": current_code,
        "test_code": adversarial_test_code,
        "passed": overall_passed,
        "failure_summary": combined_failure_summary,
        "root_cause": state.get("root_cause", ""),        # from previous diagnose_failure
        "failure_category": state.get("failure_category", ""),
        "repair_strategy": state.get("repair_strategy", ""),
    }
    # Append to history (not replace) — we want the full timeline across all iterations
    history = list(state.get("iteration_history", []))
    history.append(iteration_record)

    return {
        "last_execution_passed": overall_passed,
        "last_failure_summary": combined_failure_summary,
        "iteration_history": history,
        "status": new_status,
        "events": events,
    }