feat: Add test-suite validation to Qiskit IVR example

docs/examples/instruct_validate_repair/qiskit_code_validation/README.md

@@ -1,6 +1,6 @@
 # Qiskit Code Validation with Instruct-Validate-Repair
 
-This example demonstrates using Mellea's Instruct-Validate-Repair (IVR) pattern to generate Qiskit quantum computing code that automatically passes `flake8-qiskit-migration` validation rules (QKT rules).
+This example demonstrates using Mellea's Instruct-Validate-Repair (IVR) pattern to generate Qiskit quantum computing code that automatically passes `flake8-qiskit-migration` validation rules (QKT rules), and, optionally, the problem's own test suite.
 
 ## What This Example Does
 
@@ -9,6 +9,8 @@ Takes a prompt containing deprecated Qiskit code and:
 2. Validates the output against QKT rules
 3. Automatically repairs the code if validation fails (up to 10 attempts)
 
+`validation_helpers.py` also provides a test-suite validator for benchmark problems (Qiskit HumanEval, QuantumKatas); see [Validating with Test Suites](#validating-with-test-suites).
+
 ## Quick Start
 
 ```bash
@@ -64,7 +66,7 @@ Benchmarks on `mistral-small-3.2-24b-qiskit` model:
 
 ```
 qiskit_code_validation/
-├── qiskit_code_validation.py   # Main example
+├── qiskit_code_validation.py   # Main example (single prompt)
 ├── validation_helpers.py       # Validation utilities
 └── README.md                   # This file
 ```
@@ -73,6 +75,9 @@ qiskit_code_validation/
 - `extract_code_from_markdown()`: Extracts code from markdown blocks
 - `validate_qiskit_migration()`: Validates against QKT rules
 - `validate_input_code()`: Pre-validates input prompts
+- `validate_correctness()`: Runs code against a benchmark problem's own test suite
+
+All validators return `(is_valid, error_message)`, so the message is fed straight back to the model as the repair reason.
 
 ## Trying Different Prompts
 
@@ -199,6 +204,81 @@ qc.measure_all()
 
 **Note**: The exact output may vary depending on the model and its interpretation of the prompt.
 
+## Validating with Test Suites
+
+Passing the QKT rules only shows that the code uses current Qiskit APIs, not that it solves the task. For benchmark problems, `validate_correctness()` runs the code against the problem's own tests (`check(candidate)` for Qiskit HumanEval, `test_<entry_point>()` for QuantumKatas, with a best-effort 30s time limit) and feeds the assertion message or traceback back to the model as the repair reason.
+
+To try it, drop the snippet below into `qiskit_code_validation.py`: the imports and `passes_tests()` at the top, the last two statements in `test_qiskit_code_validation()` in place of the existing `generate_validated_qiskit_code()` call, so they reuse its `m` and `strategy`. It reads a dataset downloaded as in [Setup](#setup).
+
+```python
+import json
+
+from validation_helpers import (
+    extract_code_from_markdown,
+    validate_correctness,
+)
+
+from mellea.stdlib.requirements import req, simple_validate
+
+
+def passes_tests(problem: dict):
+    """Pass the problem's own test suite."""
+
+    def check(output: str) -> tuple[bool, str]:
+        return validate_correctness(problem, extract_code_from_markdown(output))
+
+    return req(
+        "The code must pass the problem's test suite",
+        validation_fn=simple_validate(check),
+    )
+
+
+# One dataset row = one problem dict
+problem = json.loads(next(open("data/qiskit_humaneval_hard.jsonl")))
+
+code, success, attempts = generate_validated_qiskit_code(
+    m, problem["prompt"], strategy, extra_requirements=[passes_tests(problem)]
+)
+```
+
+> ⚠️ **Security**: the test-suite validator runs `exec()` on model-generated code and on the dataset's test code, in the current process and **not** in a sandbox. Run it only on datasets you trust, on a machine where running arbitrary Python is acceptable.
+>
+> The 30s time limit is best effort: it is only checked between Python bytecodes, so code blocked inside a C call (a simulator run, `time.sleep`) is stopped only once that call returns, and a bare `except:` in the generated code can swallow it.
+
+### Setup
+
+The benchmark dependencies and the datasets are **not** part of this repository. Install and download them anywhere you like; the paths below are only examples.
+
+#### Python Dependencies
+
+The benchmark tests import Qiskit and a few plotting and graph libraries. Install them in the project environment:
+
+```bash
+uv pip install flake8-qiskit-migration "qiskit>=2.1.0,<2.5.0" qiskit-aer qiskit-ibm-runtime qiskit-ibm-transpiler matplotlib networkx pylatexenc seaborn
+```
+
+#### Datasets
+
+The benchmarks are published by Qiskit on Hugging Face, with the same fields (`task_id`, `prompt`, `canonical_solution`, `test`, `entry_point`); each row is one `problem` for `validate_correctness()`.
+
+| Benchmark | Tasks | Prompt style | License |
+|-----------|-------|--------------|---------|
+| [Qiskit HumanEval hard](https://huggingface.co/datasets/Qiskit/qiskit_humaneval_hard) | 151 | Natural-language instruction: the model writes the whole function | Apache-2.0 |
+| [Qiskit HumanEval](https://huggingface.co/datasets/Qiskit/qiskit_humaneval) | 151 | Code stub with a docstring | Apache-2.0 |
+| [Qiskit QuantumKatas](https://huggingface.co/datasets/Qiskit/Qiskit-QuantumKatas) | 350 | Task description plus a function stub | **CC BY-NC-SA 4.0 (non-commercial)** |
+
+```bash
+mkdir -p data
+# Qiskit HumanEval hard (and, the same way, Qiskit/qiskit_humaneval)
+uv run --with datasets python -c "from datasets import load_dataset; load_dataset('Qiskit/qiskit_humaneval_hard', split='test').to_json('data/qiskit_humaneval_hard.jsonl', force_ascii=False)"
+# Qiskit QuantumKatas
+curl -L -o data/qiskit_quantumkatas.jsonl https://huggingface.co/datasets/Qiskit/Qiskit-QuantumKatas/resolve/main/qiskit_quantumkatas.jsonl
+```
+
+Check each dataset's license before redistributing it, or any results generated from it: results generated from QuantumKatas fall under CC BY-NC-SA 4.0.
+
+Some Qiskit HumanEval tasks call IBM Quantum services: they can only pass with a saved IBM Quantum account (`QiskitRuntimeService.save_account(...)`); without one, even their canonical solutions fail. The remaining canonical solutions, and all 350 QuantumKatas ones, pass the test-suite validator.
+
 ## Changing the Model
 
 To try a different model, edit the `model_id` variable in the `test_qiskit_code_validation()` function:

docs/examples/instruct_validate_repair/qiskit_code_validation/validation_helpers.py

@@ -2,11 +2,19 @@
 """Helper functions for Qiskit code validation.
 
 This module provides utilities for extracting code from markdown and validating
-Qiskit code against migration rules using the flake8-qiskit-migration plugin.
+Qiskit code against migration rules using the flake8-qiskit-migration plugin,
+and against a benchmark problem's own test suite.
 """
 
 import ast
+import builtins
+import ctypes
+import os
 import re
+import tempfile
+import threading
+import traceback
+from typing import Any
 
 try:
     from flake8_qiskit_migration.plugin import Plugin
@@ -108,3 +116,121 @@ def validate_input_code(prompt: str) -> tuple[bool, str]:
         return False, error_msg
 
     return True, ""
+
+
+# Time limit for running a candidate against its tests.
+TIMEOUT_SECONDS = 30
+
+
+class _Timeout(BaseException):
+    """Raised into the validating thread when the code runs too long.
+
+    A `BaseException`, unlike `TimeoutError`, so generated code that catches
+    `Exception` cannot swallow it.
+    """
+
+
+# Builtins the executed code does not get: they could escape or stall the check.
+# This is defense in depth, not a sandbox: `__import__` is kept for `import`
+# statements, so the executed code can still use any module.
+_SAFE_BUILTINS = {
+    k: v
+    for k, v in vars(builtins).items()
+    if k not in {"exec", "eval", "compile", "breakpoint", "input"}
+}
+
+
+def _run_tests(program: str, entry: str, test_name: str) -> tuple[bool, str]:
+    """Execute `program` and call its test function, stopping it after the time limit."""
+    # Mellea validates on a worker thread, so the timeout is raised via the C API.
+    tid = ctypes.c_ulong(threading.get_ident())
+    timer = threading.Timer(
+        TIMEOUT_SECONDS,
+        ctypes.pythonapi.PyThreadState_SetAsyncExc,
+        (tid, ctypes.py_object(_Timeout)),
+    )
+    try:
+        try:
+            timer.start()
+            ns: dict[str, Any] = {"__builtins__": _SAFE_BUILTINS}
+            exec(program, ns)
+            if entry not in ns:
+                return False, f"Entry point `{entry}` not defined"
+            if test_name == "check":
+                ns["check"](ns[entry])
+            else:
+                ns[test_name]()
+            return True, ""
+        finally:
+            timer.cancel()
+            timer.join()
+            # cancel() cannot recall a _Timeout the timer already raised into this
+            # thread: clear it, so it cannot surface later inside Mellea. If it is
+            # delivered first, the except below still catches it.
+            ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None)
+    except _Timeout:
+        print(f"Validation failed: Execution timed out after {TIMEOUT_SECONDS}s")
+        return False, f"Execution timed out after {TIMEOUT_SECONDS}s"
+    except SystemExit as e:
+        # Not an Exception: left uncaught, it would stop the repair loop.
+        print(f"Validation failed: the code called sys.exit({e.code!r})")
+        return False, f"The code called sys.exit({e.code!r}) instead of returning"
+    except AssertionError as e:
+        print(f"Validation failed: Test assertion failed: {e}")
+        return False, f"Test assertion failed: {e}"
+    except Exception as e:
+        print(f"{type(e).__name__}: {e}\n{traceback.format_exc()}")
+        return False, f"{type(e).__name__}: {e}\n{traceback.format_exc()}"
+
+
+def validate_correctness(problem: dict[str, Any], code: str) -> tuple[bool, str]:
+    """Run `code` against a benchmark problem's own test suite.
+
+    Adapted from the Qiskit HumanEval test harness (Copyright IBM, Apache-2.0).
+    Supports Qiskit HumanEval (`check(candidate)`) and QuantumKatas
+    (`test_<entry_point>()`).
+
+    Warning: this `exec()`s untrusted code in the current process; it is not a
+    sandbox. The time limit is best effort: it is only checked between Python
+    bytecodes, so code blocked inside a C call (a simulator run, `time.sleep`) is
+    stopped only once that call returns, and a bare `except:` in the code can
+    swallow it.
+
+    Args:
+        problem: Problem dict with `prompt`, `test` and `entry_point` keys
+        code: Candidate solution
+
+    Returns:
+        Tuple of (is_valid, error_message)
+    """
+    # Code-stub prompts are prepended; natural-language prompts (QHE hard) are not.
+    prompt = problem["prompt"]
+    is_stub = prompt.lstrip().startswith(("from ", "import ", "def ", "#"))
+    entry = problem["entry_point"]
+    program = f"{prompt if is_stub else ''}\n{code}\n{problem['test']}"
+
+    # Call the test function the test code defines, so a candidate that defines its
+    # own `check()` cannot stand in for the real test.
+    test_fns = {
+        node.name
+        for node in ast.parse(problem["test"]).body
+        if isinstance(node, ast.FunctionDef)
+    }
+    test_name = "check" if "check" in test_fns else f"test_{entry}"
+    if test_name not in test_fns:
+        return False, f"No check() or test_{entry}() in the test code"
+
+    # Run in a scratch directory (some tasks write files) with a non-interactive
+    # matplotlib backend, and restore both afterwards.
+    cwd, mpl_backend = os.getcwd(), os.environ.get("MPLBACKEND")
+    os.environ["MPLBACKEND"] = "Agg"
+    with tempfile.TemporaryDirectory() as tmp:
+        os.chdir(tmp)
+        try:
+            return _run_tests(program, entry, test_name)
+        finally:
+            os.chdir(cwd)
+            if mpl_backend is None:
+                os.environ.pop("MPLBACKEND", None)
+            else:
+                os.environ["MPLBACKEND"] = mpl_backend