Kapso search strategies: exploring the solution space

search_strategy:
  type: "generic"
  params:
    idea_generation_model: "claude-opus-5"
    implementation_model: "claude-opus-5"
    auth_mode: "oauth"
    gate_failure_policy: "warn"
    ideation_gates: ["research", "experiment_history", "repo_memory", "leeroopedia"]
    implementation_gates: ["research", "repo_memory", "leeroopedia"]

from kapso.execution.search_strategies import SearchStrategyFactory

strategy = SearchStrategyFactory.create(
    strategy_type="generic",
    problem_handler=handler,
    llm=llm,
    coding_agent_config=config,
    params={"ideation_timeout": 300},
)

strategy.run(context, budget_progress=0.0)
best = strategy.get_best_experiment()

@dataclass
class SearchNode:
    node_id: int
    parent_node_id: Optional[int] = None
    
    # Step 1: Solution generation
    solution: str = ""
    
    # Step 2: Implementation
    branch_name: str = ""
    code_changes_summary: str = ""
    
    # Step 3: Evaluation
    evaluation_script_path: str = ""
    evaluation_output: str = ""
    
    # Step 4: Feedback
    feedback: str = ""
    score: Optional[float] = None
    should_stop: bool = False
    evaluation_valid: bool = True
    
    # Metadata
    had_error: bool = False
    error_message: str = ""
    workspace_dir: str = ""

from kapso.execution.search_strategies.base import SearchStrategy
from kapso.execution.search_strategies.factory import register_strategy

@register_strategy("my_custom_search")
class MyCustomSearch(SearchStrategy):
    def __init__(self, config, workspace_dir=None, import_from_checkpoint=False):
        # The factory passes all three — omitting the last is a TypeError
        super().__init__(config, workspace_dir, import_from_checkpoint)
        # Custom initialization

    def run(self, context, budget_progress=0.0):
        # Generate and run experiments
        solution = self.generate_solution(context)
        result = self._implement(
            solution, context,
            branch_name="experiment_0",
        )
        self.node_history.append(result)

    def get_experiment_history(self, best_last=False):
        if best_last:
            return sorted(self.node_history, key=lambda x: x.score)
        return self.node_history

    def get_best_experiment(self):
        valid = [e for e in self.node_history if not e.had_error]
        return max(valid, key=lambda x: x.score) if valid else None

    def checkout_to_best_experiment_branch(self):
        best = self.get_best_experiment()
        if best:
            self.workspace.switch_branch(best.branch_name)