""" 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, }