Framework Architecture

┌─────────────────────────────────────────────────────────────┐
│                         Runner                               │
│  (Orchestrates lifecycle, checkpointing, monitoring)         │
└────────────────┬────────────────────────────────┬────────────┘
                 │                                │
        ┌────────▼────────┐              ┌────────▼─────────┐
        │ Discovery       │              │   Evaluator      │
        │ Controller      │              │                  │
        └────────┬────────┘              └──────────────────┘
                 │
        ┌────────▼────────┐
        │ Program         │
        │ Database        │
        └─────────────────┘

class Runner:
    def __init__(
        self,
        evaluation_file: str,
        initial_program_path: Optional[str] = None,
        config_path: Optional[str] = None,
        config: Optional[Config] = None,
        output_dir: Optional[str] = None,
    )

class DiscoveryController:
    def __init__(self, controller_input: DiscoveryControllerInput):
        self.config = controller_input.config
        self.database = controller_input.database
        self.evaluator = Evaluator(config.evaluator)
        self.llms = LLMManager(config.llm)
        self.prompt_builder = PromptBuilder(config.prompt)

async def run_discovery(
    self,
    start_iteration: int,
    max_iterations: int,
    checkpoint_callback=None,
):
    for iteration in range(start_iteration, start_iteration + max_iterations):
        if self.shutdown_event.is_set():
            break

        # Core iteration
        result = await self._run_iteration(iteration)

        if result.error:
            continue

        # Store to database
        self._process_iteration_result(
            result, iteration, checkpoint_callback
        )

class ProgramDatabase(ABC):
    def __init__(self, name: str, config: DatabaseConfig, **kwargs):
        self.programs: Dict[str, Program] = {}  # All programs
        self.best_program_id: Optional[str] = None
        self.last_iteration: int = 0

    @abstractmethod
    def add(self, program: Program, iteration: Optional[int] = None) -> str:
        """Store a program and update best tracking."""
        ...

    @abstractmethod
    def sample(
        self, num_context_programs: Optional[int] = 4
    ) -> Tuple[Program, List[Program]]:
        """Select parent and context programs for next iteration."""
        ...

class Evaluator:
    def __init__(
        self,
        config: EvaluatorConfig,
        llm_judge: Optional[LLMJudge] = None,
        max_concurrent: int = 4,
    ):
        self.evaluation_file = config.evaluation_file
        self.program_suffix = config.file_suffix
        self._load_evaluation_function()

async def evaluate_program(
    self, program_solution: str, program_id: str = ""
) -> EvaluationResult:
    # 1. Write program to temp file
    with tempfile.NamedTemporaryFile(suffix=self.program_suffix) as f:
        f.write(program_solution.encode("utf-8"))
        temp_path = f.name

    # 2. Run evaluation function with timeout
    result = await asyncio.wait_for(
        loop.run_in_executor(None, self.evaluate_function, temp_path),
        timeout=self.config.timeout,
    )

    # 3. Normalize to EvaluationResult
    eval_result = self._normalize_result(result)

    # 4. Optional: add LLM judge feedback
    if self.llm_judge:
        llm_result = await self.llm_judge.evaluate(program_solution)
        eval_result.metrics.update(llm_result.metrics)

    return eval_result

@dataclass
class Program:
    # Identity
    id: str                          # UUID
    solution: str                    # Source code or prompt text
    language: str = "python"

    # Performance
    metrics: Dict[str, Any]          # {"combined_score": float, ...}

    # Lineage
    iteration_found: int = 0
    parent_id: Optional[str] = None
    other_context_ids: Optional[List[str]] = None

    # Metadata
    artifacts: Dict[str, Any]        # Feedback for LLM
    metadata: Dict[str, Any]         # Algorithm-specific data
    prompts: Optional[Dict]          # Prompt used to generate
    generation: int = 0              # Distance from initial program
    timestamp: float

@dataclass
class EvaluationResult:
    metrics: Dict[str, Any]    # Numeric scores
    artifacts: Dict[str, Any]  # Feedback text, error messages, etc.

max_iterations: 100

llm:
  models:
    - name: "gpt-4"
      weight: 1.0

search:
  type: "adaevolve"
  database:
    num_islands: 5
    migration_interval: 10

evaluator:
  timeout: 300
  cascade_evaluation: true
  cascade_thresholds: [0.7]

prompt:
  system_message: "You are an expert at optimizing algorithms."

monitor:
  enabled: true
  port: 8080

# Simple algorithms (use default controller)
register_database("topk", TopKDatabase)
register_database("best_of_n", BestOfNDatabase)
register_database("beam_search", BeamSearchDatabase)

# Advanced algorithms (custom controller)
register_database("adaevolve", AdaEvolveDatabase)
register_controller("adaevolve", AdaEvolveController)

register_database("gepa_native", GEPANativeDatabase)
register_controller("gepa_native", GEPANativeController)

register_controller("evox", CoEvolutionController)
register_database("evox_meta", SearchStrategyDatabase)

checkpoints/
  checkpoint_50/
    programs.json          # All programs with metrics
    best_program.py        # Best solution code
    best_program_info.json # Metadata about best
    prompts/               # Optional prompt logs
      <program_id>.json

skydiscover-run evaluator.py --checkpoint checkpoints/checkpoint_50

# In Runner._start_monitor()
from skydiscover.extras.monitor import MonitorServer

server = MonitorServer(host="0.0.0.0", port=8080)
server.start()

# Callback pushes programs to frontend
def monitor_callback(program: Program, iteration: int):
    server.push_event({
        "type": "program_added",
        "program_id": program.id,
        "score": program.metrics.get("combined_score"),
        "iteration": iteration,
    })