jester1177/cloudnative-devops-debug-env

1"""Heuristic baseline runner for the /baseline endpoint.2 3Applies expected_fixes directly to verify the environment + grader work e2e.4By default runs ALL scenarios of ALL tasks for deterministic, reproducible evaluation.5"""6 7 8from typing import List, Optional9 10from server.environment import CloudNativeDebugEnvironment11from server.graders import run_grader12from server.models import Action, ActionType, FileEdit, GraderResult13from server.tasks.task_registry import TASK_REGISTRY14 15 16def _heuristic_episode(env: CloudNativeDebugEnvironment, task_id: str, scenario_id: Optional[str] = None) -> GraderResult:17    """Run one episode using a heuristic that applies expected fixes."""18    obs = env.reset(task_id=task_id, scenario_id=scenario_id)19 20    # Apply each expected fix as an edit_file action21    for fix in env.expected_fixes:22        if env.done:23            break24        file_path = fix["file"]25        if file_path not in env.current_files:26            # For fixes that require creating a new file (e.g. ConfigMap),27            # create it with the expected content28            if fix["type"] == "contains":29                action = Action(30                    action_type=ActionType.EDIT_FILE,31                    edits=[FileEdit(32                        file_path=file_path,33                        new_content=fix["expected"],34                    )],35                )36                env.step(action)37            continue38 39        current_content = env.current_files[file_path].content40 41        if fix["type"] == "contains":42            # Need to ensure expected string is present43            if fix["expected"] not in current_content:44                # Try to find the broken line using hint45                hint_text = fix.get("hint", "")46                # Use edit_file with old/new content based on the fix47                # We look at original files to find what changed48                original_content = env.original_files.get(file_path)49                if original_content:50                    lines = current_content.split("\n")51                    expected = fix["expected"]52                    line_num = fix.get("line")53 54                    if line_num and 1 <= line_num <= len(lines):55                        old_line = lines[line_num - 1]56                        action = Action(57                            action_type=ActionType.REPLACE_LINE,58                            edits=[FileEdit(59                                file_path=file_path,60                                line_number=line_num,61                                new_content=expected,62                            )],63                        )64                    else:65                        # Find the line with highest character overlap to expected66                        best_line = None67                        best_idx = None68                        best_score = 069                        for i, line in enumerate(lines):70                            stripped = line.strip()71                            exp_stripped = expected.strip()72                            if not stripped or not exp_stripped:73                                continue74                            overlap = len(set(stripped) & set(exp_stripped))75                            # Use ratio of overlap to max length for scoring76                            score = overlap / max(len(exp_stripped), len(stripped))77                            if score > 0.5 and score > best_score:78                                best_line = line79                                best_idx = i80                                best_score = score81 82                        if best_line is not None:83                            action = Action(84                                action_type=ActionType.EDIT_FILE,85                                edits=[FileEdit(86                                    file_path=file_path,87                                    old_content=best_line,88                                    new_content=expected,89                                )],90                            )91                        else:92                            # Append the expected content93                            action = Action(94                                action_type=ActionType.ADD_LINE,95                                edits=[FileEdit(96                                    file_path=file_path,97                                    new_content=expected,98                                )],99                            )100                    env.step(action)101 102        elif fix["type"] == "not_contains":103            # Need to ensure expected string is NOT present104            if fix["expected"] in current_content:105                action = Action(106                    action_type=ActionType.DELETE_BLOCK,107                    edits=[FileEdit(108                        file_path=file_path,109                        old_content=fix["expected"],110                    )],111                )112                env.step(action)113 114        elif fix["type"] == "line_equals":115            line_num = int(fix.get("line", 0))116            if line_num >= 1:117                action = Action(118                    action_type=ActionType.REPLACE_LINE,119                    edits=[FileEdit(120                        file_path=file_path,121                        line_number=line_num,122                        new_content=str(fix["expected"]),123                    )],124                )125                env.step(action)126 127    # Submit if not already done128    if not env.done:129        env.step(Action(action_type=ActionType.SUBMIT))130 131    return run_grader(task_id, env.trajectory)132 133 134def run_baseline_episodes(task_id: Optional[str] = None, num_episodes: Optional[int] = None) -> List[GraderResult]:135    """Run baseline episodes across tasks.136 137    Args:138        task_id: Specific task to run, or None for all tasks.139        num_episodes: Max scenarios per task. None = run ALL scenarios (default).140 141    Returns:142        List of GraderResult for each episode.143    """144    task_ids: List[str]145    if task_id:146        if task_id not in TASK_REGISTRY:147            raise ValueError(f"Unknown task: {task_id}")148        task_ids = [task_id]149    else:150        task_ids = list(TASK_REGISTRY.keys())151 152    results: List[GraderResult] = []153    for tid in task_ids:154        task_cls = TASK_REGISTRY[tid]155        scenarios = task_cls.SCENARIOS156        for idx, scenario in enumerate(scenarios):157            if num_episodes is not None and idx >= num_episodes:158                break159            env = CloudNativeDebugEnvironment()160            result = _heuristic_episode(env, tid, scenario["id"])161            results.append(result)162 163    return results164