jester1177/cloud-native-debug-env
0
1"""Core environment logic."""2 3import copy4import random5from typing import Any, Dict, List, Optional, Tuple6 7from server.models import (8 Action,9 ActionType,10 ErrorInfo,11 ErrorPhase,12 FileContent,13 FileType,14 Observation,15 TaskDifficulty,16)17from server.simulators.docker_simulator import DockerSimulator18from server.simulators.k8s_simulator import KubernetesSimulator19from server.simulators.workflow_simulator import WorkflowSimulator20from server.tasks.task_registry import TASK_REGISTRY, get_task21 22 23class CICDDebugEnvironment:24 MAX_STEPS = 1025 MAX_HINTS = 326 27 def _resolve_task_id(self, task_id: Optional[object]) -> Optional[str]:28 """Resolve task identifiers from either canonical string IDs or numeric indices.29 30 Supports:31 - canonical ids (e.g. "dockerfile_syntax")32 - numeric values (1-based preferred, 0-based tolerated)33 - numeric strings ("1", "0")34 """35 if task_id is None:36 return None37 38 keys = list(TASK_REGISTRY.keys())39 40 if isinstance(task_id, int):41 if 1 <= task_id <= len(keys):42 return keys[task_id - 1]43 if 0 <= task_id < len(keys):44 return keys[task_id]45 return str(task_id)46 47 if isinstance(task_id, str):48 raw = task_id.strip()49 if raw in TASK_REGISTRY:50 return raw51 if raw.isdigit():52 numeric = int(raw)53 if 1 <= numeric <= len(keys):54 return keys[numeric - 1]55 if 0 <= numeric < len(keys):56 return keys[numeric]57 return raw58 lowered = raw.lower()59 for key in keys:60 if key.lower() == lowered:61 return key62 return raw63 64 return str(task_id)65 66 def _find_workflow_file(self) -> Optional[FileContent]:67 """Return the first workflow file found in current_files."""68 for path, fc in self.current_files.items():69 if path.startswith(".github/workflows/") and path.endswith(".yml"):70 return fc71 return None72 73 def _validation_snapshot(self) -> Dict[str, bool]:74 docker_result = self.docker_sim.validate(self.current_files.get("Dockerfile"), self.current_files)75 workflow_file = self._find_workflow_file()76 workflow_result = self.workflow_sim.validate(workflow_file, self.current_files)77 k8s_result = self.k8s_sim.validate(self.current_files)78 return {79 "docker_build_valid": bool(docker_result.get("build_success", False)),80 "workflow_parse_valid": bool(workflow_result.get("parse_success", False)),81 "k8s_valid": bool(k8s_result.get("valid", True)),82 }83 84 def __init__(self):85 self.docker_sim = DockerSimulator()86 self.workflow_sim = WorkflowSimulator()87 self.k8s_sim = KubernetesSimulator()88 89 self.current_task_id: Optional[str] = None90 self.current_scenario_id: Optional[str] = None91 self.current_difficulty: Optional[TaskDifficulty] = None92 self.current_task = None93 94 self.original_files: Dict[str, FileContent] = {}95 self.current_files: Dict[str, FileContent] = {}96 self.expected_fixes: List[Dict[str, Any]] = []97 98 self.current_error: Optional[ErrorInfo] = None99 self.issues_total = 0100 self.issues_fixed = 0101 102 self.step_count = 0103 self.episode_reward = 0.0104 self.done = False105 self.hints_used = 0106 107 self.trajectory: List[Dict[str, Any]] = []108 self.last_action_success: Optional[bool] = None109 self.last_action_feedback: Optional[str] = None110 111 def reset(self, task_id: Optional[str] = None, scenario_id: Optional[str] = None, seed: Optional[int] = None):112 if seed is not None:113 random.seed(seed)114 115 task_id = self._resolve_task_id(task_id)116 117 if task_id is None:118 task_id = random.choice(list(TASK_REGISTRY.keys()))119 if task_id not in TASK_REGISTRY:120 raise ValueError(f"Unknown task: {task_id}")121 122 self.current_task_id = task_id123 self.current_task = get_task(task_id)124 self.current_difficulty = self.current_task.DIFFICULTY125 126 scenario = self.current_task.load_scenario(scenario_id)127 self.current_scenario_id = scenario["id"]128 129 self.original_files = {130 f["path"]: FileContent(131 path=f["path"],132 content=f["content"],133 file_type=FileType(f["type"]),134 line_count=f["content"].count("\n") + 1,135 )136 for f in scenario["files"]137 }138 self.current_files = copy.deepcopy(self.original_files)139 140 self.current_error = ErrorInfo(141 phase=ErrorPhase(scenario["error"]["phase"]),142 error_message=scenario["error"]["message"],143 exit_code=scenario["error"].get("exit_code"),144 failed_step=scenario["error"].get("failed_step"),145 line_hint=scenario["error"].get("line_hint"),146 )147 148 self.expected_fixes = scenario["expected_fixes"]149 self.issues_total = len(self.expected_fixes)150 self.issues_fixed = 0151 152 self.step_count = 0153 self.episode_reward = 0.0154 self.done = False155 self.hints_used = 0156 self.trajectory = []157 self.last_action_success = None158 self.last_action_feedback = None159 160 return self.get_observation()161 162 def step(self, action: Action) -> Tuple[Observation, float, bool, Dict[str, Any]]:163 if self.done:164 return self.get_observation(), 0.0, True, {"error": "Episode already done"}165 166 self.step_count += 1167 168 if action.action_type == ActionType.REQUEST_HINT:169 reward, feedback = self._handle_hint_request()170 elif action.action_type == ActionType.SUBMIT:171 reward, feedback = self._handle_submit()172 else:173 reward, feedback = self._handle_edit(action)174 175 self.last_action_feedback = feedback176 self.episode_reward += reward177 178 info: Dict[str, Any] = {}179 if self.step_count >= self.MAX_STEPS:180 self.done = True181 info["termination_reason"] = "max_steps"182 elif action.action_type == ActionType.SUBMIT:183 self.done = True184 info["termination_reason"] = "submitted"185 elif self.issues_fixed == self.issues_total:186 self.done = True187 info["termination_reason"] = "all_fixed"188 189 info["issues_fixed"] = self.issues_fixed190 info["issues_total"] = self.issues_total191 192 self.trajectory.append(193 {"step": self.step_count, "action": action.model_dump(), "reward": reward, "done": self.done, "info": info}194 )195 196 return self.get_observation(), reward, self.done, info197 198 def _handle_edit(self, action: Action) -> Tuple[float, str]:199 if not action.edits:200 self.last_action_success = False201 return 0.0, "No edits provided"202 203 before_validation = self._validation_snapshot()204 205 reward = 0.0206 feedbacks: List[str] = []207 applied_count = 0208 for edit in action.edits:209 if edit.file_path not in self.current_files:210 # Allow creating new files (needed for K8s ConfigMap scenarios etc.)211 if action.action_type == ActionType.EDIT_FILE and edit.new_content:212 ft = FileType.OTHER213 if edit.file_path.startswith("k8s/") or edit.file_path.endswith(".yaml") or edit.file_path.endswith(".yml"):214 ft = FileType.KUBERNETES215 self.current_files[edit.file_path] = FileContent(216 path=edit.file_path,217 content=edit.new_content,218 file_type=ft,219 line_count=edit.new_content.count("\n") + 1,220 )221 feedbacks.append(f"Created new file: {edit.file_path}")222 applied_count += 1223 continue224 feedbacks.append(f"File not found: {edit.file_path}")225 continue226 227 file_content = self.current_files[edit.file_path]228 lines = file_content.content.split("\n")229 230 if action.action_type == ActionType.REPLACE_LINE:231 if edit.line_number and 1 <= edit.line_number <= len(lines):232 lines[edit.line_number - 1] = edit.new_content or ""233 feedbacks.append(f"Replaced line {edit.line_number} in {edit.file_path}")234 applied_count += 1235 else:236 feedbacks.append(f"Invalid line number: {edit.line_number}")237 continue238 elif action.action_type == ActionType.ADD_LINE:239 insert_at = edit.line_number - 1 if edit.line_number else len(lines)240 insert_at = max(0, min(insert_at, len(lines)))241 lines.insert(insert_at, edit.new_content or "")242 feedbacks.append(f"Added line at {insert_at + 1} in {edit.file_path}")243 applied_count += 1244 elif action.action_type == ActionType.DELETE_LINE:245 if edit.line_number and 1 <= edit.line_number <= len(lines):246 del lines[edit.line_number - 1]247 feedbacks.append(f"Deleted line {edit.line_number} in {edit.file_path}")248 applied_count += 1249 else:250 feedbacks.append(f"Invalid line number: {edit.line_number}")251 continue252 elif action.action_type == ActionType.EDIT_FILE:253 if edit.old_content and edit.old_content in file_content.content:254 new_content = file_content.content.replace(edit.old_content, edit.new_content or "", 1)255 lines = new_content.split("\n")256 feedbacks.append(f"Edited content in {edit.file_path}")257 applied_count += 1258 else:259 feedbacks.append(f"Content not found in {edit.file_path}")260 continue261 elif action.action_type == ActionType.ADD_BLOCK:262 insert_at = edit.line_number - 1 if edit.line_number else len(lines)263 insert_at = max(0, min(insert_at, len(lines)))264 block_lines = (edit.new_content or "").split("\n")265 lines[insert_at:insert_at] = block_lines266 feedbacks.append(f"Added block at {insert_at + 1} in {edit.file_path}")267 applied_count += 1268 elif action.action_type == ActionType.DELETE_BLOCK:269 if edit.old_content and edit.old_content in file_content.content:270 new_content = file_content.content.replace(edit.old_content, "", 1)271 lines = new_content.split("\n")272 feedbacks.append(f"Deleted block in {edit.file_path}")273 applied_count += 1274 else:275 feedbacks.append(f"Block content not found in {edit.file_path}")276 continue277 else:278 feedbacks.append(f"Unsupported action type for now: {action.action_type.value}")279 continue280 281 new_content = "\n".join(lines)282 self.current_files[edit.file_path] = FileContent(283 path=file_content.path,284 content=new_content,285 file_type=file_content.file_type,286 line_count=len(lines),287 )288 289 reward += self._check_fix_progress()290 291 after_validation = self._validation_snapshot()292 if not before_validation["docker_build_valid"] and after_validation["docker_build_valid"]:293 reward += 0.1294 feedbacks.append("Docker build validity improved")295 if not before_validation["workflow_parse_valid"] and after_validation["workflow_parse_valid"]:296 reward += 0.1297 feedbacks.append("Workflow parse validity improved")298 if not before_validation["k8s_valid"] and after_validation["k8s_valid"]:299 reward += 0.1300 feedbacks.append("Kubernetes manifest validity improved")301 302 if applied_count == 0:303 self.last_action_success = False304 return max(-0.02, reward - 0.02), "; ".join(feedbacks) or "No edit applied"305 306 self.last_action_success = True307 return max(0.0, reward), "; ".join(feedbacks)308 309 def _check_fix_progress(self) -> float:310 fixes_applied = 0311 for fix in self.expected_fixes:312 file_path = fix["file"]313 if file_path not in self.current_files:314 # For "contains" checks on missing files, the fix is not applied315 # For "not_contains" checks on missing files, consider it fixed316 if fix["type"] == "not_contains":317 fixes_applied += 1318 continue319 current_content = self.current_files[file_path].content320 if fix["type"] == "contains" and fix["expected"] in current_content:321 fixes_applied += 1322 if fix["type"] == "not_contains" and fix["expected"] not in current_content:323 fixes_applied += 1324 if fix["type"] == "line_equals":325 lines = current_content.split("\n")326 line_num = int(fix.get("line", 0))327 if 1 <= line_num <= len(lines):328 if lines[line_num - 1].strip() == str(fix["expected"]).strip():329 fixes_applied += 1330 331 new_fixed = fixes_applied - self.issues_fixed332 if new_fixed > 0:333 self.issues_fixed = fixes_applied334 return 0.3 * new_fixed335 return 0.0336 337 def _handle_submit(self) -> Tuple[float, str]:338 docker_result = self.docker_sim.validate(self.current_files.get("Dockerfile"), self.current_files)339 workflow_file = self._find_workflow_file()340 workflow_result = self.workflow_sim.validate(workflow_file, self.current_files)341 k8s_result = self.k8s_sim.validate(self.current_files)342 343 has_k8s = any(fc.file_type == FileType.KUBERNETES for fc in self.current_files.values())344 has_docker = "Dockerfile" in self.current_files345 has_workflow = workflow_file is not None346 347 reward = 0.0348 parts: List[str] = []349 350 # Determine weight distribution based on what file types are present351 if has_docker and has_workflow and has_k8s:352 # Full stack: Docker 20%, Workflow 30%, K8s 30%, fix progress 20%353 docker_w, wf_w, k8s_w = 0.20, 0.30, 0.30354 elif has_docker and has_workflow:355 docker_w, wf_w, k8s_w = 0.50, 0.50, 0.0356 elif has_docker and has_k8s:357 docker_w, wf_w, k8s_w = 0.40, 0.0, 0.40358 elif has_workflow and has_k8s:359 docker_w, wf_w, k8s_w = 0.0, 0.40, 0.40360 elif has_k8s:361 docker_w, wf_w, k8s_w = 0.0, 0.0, 0.80362 elif has_docker:363 docker_w, wf_w, k8s_w = 0.50, 0.0, 0.0364 else:365 docker_w, wf_w, k8s_w = 0.0, 0.50, 0.0366 367 # Docker validation368 if has_docker:369 if docker_result.get("build_success"):370 reward += docker_w * 0.6371 parts.append("Docker build: PASS")372 else:373 parts.append(f"Docker build: FAIL - {docker_result.get('error', 'unknown')}")374 375 if docker_result.get("run_success"):376 reward += docker_w * 0.4377 parts.append("Docker run: PASS")378 else:379 parts.append(f"Docker run: FAIL - {docker_result.get('run_error', 'unknown')}")380 381 # Workflow validation382 if has_workflow:383 if workflow_result["parse_success"]:384 reward += wf_w * 0.4385 parts.append("Workflow parse: PASS")386 else:387 parts.append(f"Workflow parse: FAIL - {workflow_result.get('error', 'unknown')}")388 389 if workflow_result["execution_success"]:390 reward += wf_w * 0.6391 parts.append("Workflow execution: PASS")392 else:393 parts.append(f"Workflow execution: FAIL - {workflow_result.get('exec_error', 'unknown')}")394 395 # Kubernetes validation396 if has_k8s:397 if k8s_result["valid"]:398 reward += k8s_w * 0.4399 parts.append("K8s manifests: VALID")400 else:401 k8s_errors = k8s_result.get("errors", [])402 parts.append(f"K8s manifests: INVALID - {'; '.join(k8s_errors[:2])}")403 404 pod_status = k8s_result.get("pod_status", "N/A")405 if pod_status == "Running":406 reward += k8s_w * 0.3407 parts.append(f"K8s pod status: {pod_status}")408 else:409 parts.append(f"K8s pod status: {pod_status}")410 411 svc_status = k8s_result.get("service_status", "N/A")412 if "active" in svc_status.lower() or svc_status == "N/A":413 reward += k8s_w * 0.3414 parts.append(f"K8s service: {svc_status}")415 else:416 parts.append(f"K8s service: {svc_status}")417 418 self.last_action_success = reward >= 0.6419 return reward, "; ".join(parts)420 421 def _handle_hint_request(self) -> Tuple[float, str]:422 if self.hints_used >= self.MAX_HINTS:423 self.last_action_success = False424 return 0.0, "No hints remaining"425 426 self.hints_used += 1427 for fix in self.expected_fixes:428 file_path = fix["file"]429 if file_path not in self.current_files:430 continue431 current_content = self.current_files[file_path].content432 is_fixed = False433 if fix["type"] == "contains":434 is_fixed = fix["expected"] in current_content435 elif fix["type"] == "not_contains":436 is_fixed = fix["expected"] not in current_content437 if not is_fixed:438 hint = fix.get("hint", f"Check {file_path}")439 self.last_action_success = True440 return -0.05, f"Hint ({self.hints_used}/{self.MAX_HINTS}): {hint}"441 442 self.last_action_success = True443 return 0.0, "All known issues appear to be fixed"444 445 def get_observation(self) -> Observation:446 return Observation(447 task_id=self.current_task_id or "unknown",448 task_description=self.current_task.DESCRIPTION if self.current_task else "",449 difficulty=self.current_difficulty or TaskDifficulty.EASY,450 files=list(self.current_files.values()),451 error=self.current_error or ErrorInfo(452 phase=ErrorPhase.DOCKER_BUILD,453 error_message="No error",454 ),455 available_secrets=self.current_task.AVAILABLE_SECRETS if self.current_task else [],456 runner_os="ubuntu-latest",457 step_number=self.step_count,458 max_steps=self.MAX_STEPS,459 hints_used=self.hints_used,460 hints_available=self.MAX_HINTS - self.hints_used,461 last_action_success=self.last_action_success,462 last_action_feedback=self.last_action_feedback,463 issues_found=self.issues_fixed,464 issues_fixed=self.issues_fixed,465 total_issues=max(1, self.issues_total),466 )467