Postmortem: How Hallucinations in Mistral Large 2 and Codeium 1.5 Caused a 2026 Production Bug

import ast
import sys
import json
import typing
from pathlib import Path

class MigrationValidationError(Exception):
    """Custom exception for migration validation failures."""
    pass

def parse_migration_file(file_path: Path) -> ast.Module:
    """
    Parse a Python migration file into an AST tree.

    Args:
        file_path: Path to the migration file.

    Returns:
        Parsed AST module.

    Raises:
        MigrationValidationError: If the file cannot be parsed.
    """
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            source = f.read()
        return ast.parse(source, filename=str(file_path))
    except SyntaxError as e:
        raise MigrationValidationError(f"Syntax error in {file_path}: {e}") from e
    except IOError as e:
        raise MigrationValidationError(f"Failed to read {file_path}: {e}") from e

def check_foreign_key_constraints(tree: ast.Module, file_path: Path) -> list[str]:
    """
    Check for missing foreign key constraints in SQLAlchemy migration code.

    Args:
        tree: AST tree of the migration.
        file_path: Path to the migration file (for error context).

    Returns:
        List of error messages.
    """
    errors = []
    # Look for SQLAlchemy Table definitions or migration operations
    for node in ast.walk(tree):
        # Check for ForeignKey or ForeignKeyConstraint usage
        if isinstance(node, ast.Call):
            if hasattr(node.func, 'attr') and node.func.attr in ('ForeignKey', 'ForeignKeyConstraint'):
                continue
            # Check if we're in a migration that adds a column without FK
            if hasattr(node.func, 'attr') and node.func.attr == 'add_column':
                # Check if column args include a ForeignKey
                has_fk = False
                for arg in node.args:
                    if isinstance(arg, ast.Call) and hasattr(arg.func, 'attr') and arg.func.attr == 'Column':
                        for col_arg in arg.args + arg.keywords:
                            if isinstance(col_arg, ast.Call) and hasattr(col_arg.func, 'attr') and col_arg.func.attr == 'ForeignKey':
                                has_fk = True
                                break
                if not has_fk:
                    errors.append(f"Potential missing foreign key in {file_path}: add_column call without ForeignKey")
    return errors

def check_type_hints(tree: ast.Module, file_path: Path) -> list[str]:
    """
    Check for hallucinated or invalid type hints in migration code.

    Args:
        tree: AST tree of the migration.
        file_path: Path to the migration file (for error context).

    Returns:
        List of error messages.
    """
    errors = []
    valid_types = {'Dict', 'List', 'Optional', 'Tuple', 'Set', 'Any', 'Union', 'str', 'int', 'float', 'bool', 'bytes'}
    for node in ast.walk(tree):
        if isinstance(node, ast.FunctionDef):
            # Check return type annotation
            if node.returns:
                annotation_str = ast.unparse(node.returns)
                # Check for lowercase or invalid types
                if any(t not in valid_types and t[0].islower() for t in annotation_str.split() if t.isidentifier()):
                    errors.append(f"Invalid type hint in {file_path}: {annotation_str}")
    return errors

def generate_validation_report(errors: list[str], file_path: Path) -> dict:
    """
    Generate a JSON validation report for the migration.

    Args:
        errors: List of error messages.
        file_path: Path to the migration file.

    Returns:
        Report dictionary.
    """
    return {
        "file": str(file_path),
        "valid": len(errors) == 0,
        "error_count": len(errors),
        "errors": errors
    }

def main():
    if len(sys.argv) != 2:
        print(f"Usage: {sys.argv[0]} ")
        sys.exit(1)

    migration_path = Path(sys.argv[1])
    if not migration_path.exists():
        print(f"Error: {migration_path} does not exist")
        sys.exit(1)

    try:
        tree = parse_migration_file(migration_path)
        fk_errors = check_foreign_key_constraints(tree, migration_path)
        type_errors = check_type_hints(tree, migration_path)
        all_errors = fk_errors + type_errors
        report = generate_validation_report(all_errors, migration_path)
        print(json.dumps(report, indent=2))
        if not report["valid"]:
            sys.exit(1)
    except MigrationValidationError as e:
        print(f"Validation failed: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

import os
import time
import json
import csv
from typing import List, Dict, Optional
from mistralai.client import MistralClient
from mistralai.models.chat_completion import ChatMessage

# Configuration
MISTRAL_API_KEY = os.getenv("MISTRAL_API_KEY", "")
MODELS_TO_BENCHMARK = ["mistral-large-latest", "open-mistral-nemo", "codellama/CodeLlama-70b-Instruct-hf"]
BENCHMARK_PROMPTS = [
    "Generate a Python 3.12 type hint for a function that returns a dictionary with string keys and optional integer values.",
    "Write a SQLAlchemy 2.0 migration to add a 'user_id' column to the 'orders' table, linking to the 'users' table via foreign key.",
    "Generate a type hint for an async function that fetches a list of user objects from an API."
]
HALLUCINATION_THRESHOLD = 0.1  # 10% invalid responses count as hallucination

class HallucinationBenchmarker:
    """Benchmark LLM hallucination rates for code generation tasks."""

    def __init__(self, api_key: str):
        self.client = MistralClient(api_key=api_key)
        self.results: List[Dict] = []

    def check_type_hint_validity(self, response: str) -> bool:
        """
        Check if a generated type hint is valid Python 3.12 syntax.

        Args:
            response: LLM-generated type hint string.

        Returns:
            True if valid, False otherwise.
        """
        try:
            # Wrap in a function definition to parse
            test_code = f"def test() -> {response}: pass"
            compile(test_code, "", "exec")
            return True
        except SyntaxError:
            return False

    def check_migration_validity(self, response: str) -> bool:
        """
        Check if a generated migration has a foreign key constraint.

        Args:
            response: LLM-generated migration code.

        Returns:
            True if valid, False otherwise.
        """
        return "ForeignKey" in response or "ForeignKeyConstraint" in response

    def run_benchmark(self, model: str, prompts: List[str]) -> Dict:
        """
        Run benchmark for a single model.

        Args:
            model: Model name to benchmark.
            prompts: List of prompts to send.

        Returns:
            Benchmark results dictionary.
        """
        total_prompts = len(prompts)
        valid_responses = 0
        latencies = []

        for prompt in prompts:
            start_time = time.time()
            try:
                messages = [ChatMessage(role="user", content=prompt)]
                response = self.client.chat(model=model, messages=messages)
                latency = time.time() - start_time
                latencies.append(latency)

                generated_text = response.choices[0].message.content
                # Check validity based on prompt type
                if "type hint" in prompt.lower():
                    is_valid = self.check_type_hint_validity(generated_text)
                elif "migration" in prompt.lower():
                    is_valid = self.check_migration_validity(generated_text)
                else:
                    is_valid = True  # Default to valid for unknown prompt types

                if is_valid:
                    valid_responses += 1

            except Exception as e:
                print(f"Error benchmarking {model}: {e}")
                latencies.append(time.time() - start_time)

        hallucination_rate = 1 - (valid_responses / total_prompts)
        avg_latency = sum(latencies) / len(latencies) if latencies else 0

        return {
            "model": model,
            "total_prompts": total_prompts,
            "valid_responses": valid_responses,
            "hallucination_rate": round(hallucination_rate, 4),
            "avg_latency_s": round(avg_latency, 2),
            "p99_latency_s": round(sorted(latencies)[-1] if latencies else 0, 2)
        }

    def run_all_benchmarks(self) -> None:
        """Run benchmarks for all configured models."""
        if not self.client.api_key:
            raise ValueError("MISTRAL_API_KEY environment variable not set")

        for model in MODELS_TO_BENCHMARK:
            print(f"Benchmarking {model}...")
            result = self.run_benchmark(model, BENCHMARK_PROMPTS)
            self.results.append(result)
            print(f"Completed {model}: {result['hallucination_rate']*100}% hallucination rate")

    def save_results(self, output_path: str = "benchmark_results.csv") -> None:
        """
        Save benchmark results to CSV.

        Args:
            output_path: Path to output CSV file.
        """
        with open(output_path, 'w', newline='', encoding='utf-8') as f:
            writer = csv.DictWriter(f, fieldnames=self.results[0].keys())
            writer.writeheader()
            writer.writerows(self.results)

def main():
    if not MISTRAL_API_KEY:
        print("Error: Set MISTRAL_API_KEY environment variable")
        return

    benchmarker = HallucinationBenchmarker(api_key=MISTRAL_API_KEY)
    try:
        benchmarker.run_all_benchmarks()
        benchmarker.save_results()
        print("Benchmark complete. Results saved to benchmark_results.csv")
    except Exception as e:
        print(f"Benchmark failed: {e}")

if __name__ == "__main__":
    main()

import os
import time
import requests
from typing import Optional, Dict, Any, List
from pydantic import BaseModel, ValidationError, Field

# Configuration
CODEIUM_API_KEY = os.getenv("CODEIUM_API_KEY", "")
CODEIUM_API_BASE = "https://api.codeium.com/v1"
MAX_RETRIES = 3
RETRY_DELAY_S = 1

class GeneratedCodeResponse(BaseModel):
    """Schema for validated Codeium generated code response."""
    code: str = Field(..., description="Generated code string")
    language: str = Field(..., description="Programming language of generated code")
    confidence: float = Field(..., ge=0.0, le=1.0, description="Model confidence score")
    warnings: List[str] = Field(default_factory=list, description="Generation warnings")

class CodeiumValidatedClient:
    """Validated client for Codeium 1.5 API with retry and schema enforcement."""

    def __init__(self, api_key: str, base_url: str = CODEIUM_API_BASE):
        self.api_key = api_key
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        })

    def _validate_response(self, response_data: Dict[str, Any]) -> GeneratedCodeResponse:
        """
        Validate API response against Pydantic schema.

        Args:
            response_data: Raw response JSON from Codeium API.

        Returns:
            Validated response object.

        Raises:
            ValidationError: If response does not match schema.
        """
        try:
            return GeneratedCodeResponse(**response_data)
        except ValidationError as e:
            raise ValueError(f"Invalid Codeium response: {e}") from e

    def generate_code(
        self,
        prompt: str,
        language: str = "python",
        context: Optional[str] = None,
        timeout: int = 30
    ) -> GeneratedCodeResponse:
        """
        Generate code with retry logic and validation.

        Args:
            prompt: Code generation prompt.
            language: Target programming language.
            context: Optional code context (e.g., existing file contents).
            timeout: Request timeout in seconds.

        Returns:
            Validated generated code response.

        Raises:
            RuntimeError: If all retries fail.
            ValueError: If response validation fails.
        """
        endpoint = f"{self.base_url}/generate"
        payload = {
            "prompt": prompt,
            "language": language,
            "context": context,
            "model": "codeium-1.5"
        }

        last_error = None
        for attempt in range(MAX_RETRIES):
            try:
                response = self.session.post(endpoint, json=payload, timeout=timeout)
                response.raise_for_status()
                response_data = response.json()
                validated = self._validate_response(response_data)
                return validated
            except requests.exceptions.RequestException as e:
                last_error = e
                if attempt < MAX_RETRIES - 1:
                    time.sleep(RETRY_DELAY_S * (2 ** attempt))  # Exponential backoff
            except ValidationError as e:
                last_error = e
                if attempt < MAX_RETRIES - 1:
                    time.sleep(RETRY_DELAY_S)

        raise RuntimeError(f"Failed to generate code after {MAX_RETRIES} attempts: {last_error}")

    def close(self):
        """Close the requests session."""
        self.session.close()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

def main():
    if not CODEIUM_API_KEY:
        print("Error: Set CODEIUM_API_KEY environment variable")
        return

    prompt = "Generate a SQLAlchemy 2.0 migration to add a 'created_at' timestamp column to the 'users' table."

    with CodeiumValidatedClient(api_key=CODEIUM_API_KEY) as client:
        try:
            response = client.generate_code(prompt=prompt, language="python")
            print(f"Generated Code (confidence: {response.confidence}):")
            print(response.code)
            if response.warnings:
                print(f"Warnings: {response.warnings}")
        except Exception as e:
            print(f"Code generation failed: {e}")

if __name__ == "__main__":
    main()

from pydantic import BaseModel, Field
from typing import List

class MigrationOutput(BaseModel):
    code: str = Field(..., description="Valid SQLAlchemy migration code")
    has_foreign_keys: bool = Field(..., description="True if migration includes FK constraints")
    type_hints_valid: bool = Field(..., description="True if all type hints are valid Python 3.12")
    warnings: List[str] = Field(default_factory=list)

def run_custom_benchmark(model, prompts, validator):
    results = []
    for prompt in prompts:
        response = model.generate(prompt)
        is_valid = validator(response)
        results.append({"prompt": prompt, "valid": is_valid})
    return sum(r["valid"] for r in results) / len(results)

def route_llm_task(task_criticality: str):
    if task_criticality == "low":
        return "mistral-nemo"
    elif task_criticality == "medium":
        return "codeium-1.5"
    else:
        return "mistral-large"