finally added all

.gitignore

@@ -7,7 +7,7 @@ __pycache__/
 # Keep source folder env/ tracked; only ignore venv directories.
 .venv/
 venv/
-ENV/
+
 env.bak/
 venv.bak/
 

env/__init__.py

@@ -0,0 +1,4 @@
+from env.rewards import RewardCalculator
+from env.environment import CICDDebuggerEnvironment, MAX_STEPS, REQUIRED_TOOLS
+
+__all__ = ["RewardCalculator", "CICDDebuggerEnvironment", "REQUIRED_TOOLS", "MAX_STEPS"]

env/anti_hacking.py

@@ -0,0 +1,208 @@
+from __future__ import annotations
+
+import re
+from typing import Any
+
+import yaml
+
+
+class AntiHackingDetector:
+    """Detects shortcut behaviors that can game CI-fix rewards."""
+
+    STAGE_SKIP_PATTERNS = (
+        r"\bif\s*:\s*false\b",
+        r"\bwhen\s*:\s*never\b",
+        r"\bon\s*:\s*\[\s*\]\b",
+        r"\bon\s*:\s*{}",
+        r"\ballow_failure\s*:\s*true\b",
+    )
+
+    FAKE_SUCCESS_PATTERNS = (
+        r"echo\s+[\"']?tests\s+passed[\"']?",
+        r"echo\s+[\"']?success[\"']?",
+        r"\bexit\s+0\b",
+        r"\btrue\b\s*#?\s*force",
+    )
+
+    HARDCODED_OUTPUT_PATTERNS = (
+        r"\bstatus\s*:\s*success\b",
+        r"\bresult\s*:\s*passed\b",
+        r"\btests?_passed\s*=\s*true\b",
+        r"\bforce_success\b",
+    )
+
+    BROKEN_COMMAND_PATTERNS = (
+        r"\bnpm\s+tset\b",
+        r"\bpyhton\b",
+        r"\bpip\s+isntall\b",
+        r"\bgo\s+tset\b",
+    )
+
+    def penalty_stage_skipping(self, config_text: str) -> float:
+        hits = self._count_hits(config_text, self.STAGE_SKIP_PATTERNS)
+        if hits == 0:
+            return 0.0
+        return -min(0.20 * hits, 0.70)
+
+    def penalty_fake_success(self, config_text: str) -> float:
+        hits = self._count_hits(config_text, self.FAKE_SUCCESS_PATTERNS)
+        if hits == 0:
+            return 0.0
+
+        normalized = (config_text or "").lower()
+        has_real_test_cmd = any(token in normalized for token in ("npm test", "pytest", "go test", "mvn test", "yarn test", "pnpm test"))
+        base = 0.15 if has_real_test_cmd else 0.25
+        return -min(base * hits, 0.70)
+
+    def penalty_hardcoded_outputs(self, config_text: str) -> float:
+        hits = self._count_hits(config_text, self.HARDCODED_OUTPUT_PATTERNS)
+        if hits == 0:
+            return 0.0
+        return -min(0.18 * hits, 0.60)
+
+    def penalty_invalid_config(self, config_text: str) -> float:
+        if not (config_text or "").strip():
+            return -0.30
+        if not self._is_yaml_valid(config_text):
+            return -0.35
+        return 0.0
+
+    def penalty_breaking_pipeline(self, previous_config: str, new_config: str) -> float:
+        if not previous_config or not new_config:
+            return 0.0
+
+        penalty = 0.0
+
+        previous_valid = self._is_yaml_valid(previous_config)
+        new_valid = self._is_yaml_valid(new_config)
+        if previous_valid and not new_valid:
+            penalty -= 0.40
+
+        previous_stages = self._extract_stage_names(previous_config)
+        new_stages = self._extract_stage_names(new_config)
+        missing_stages = previous_stages - new_stages
+        if missing_stages:
+            penalty -= min(0.15 * len(missing_stages), 0.45)
+
+        previous_broken = self._count_hits(previous_config, self.BROKEN_COMMAND_PATTERNS)
+        new_broken = self._count_hits(new_config, self.BROKEN_COMMAND_PATTERNS)
+        if new_broken > previous_broken:
+            penalty -= min(0.10 * (new_broken - previous_broken), 0.30)
+
+        return max(-1.0, penalty)
+
+    def penalty_excessive_edits(
+        self,
+        edit_count: int | dict[str, Any] | None = None,
+        changed_files_count: int = 0,
+        changed_lines_count: int = 0,
+    ) -> float:
+        if isinstance(edit_count, dict):
+            changed_files_count = int(edit_count.get("changed_files_count", changed_files_count) or 0)
+            changed_lines_count = int(edit_count.get("changed_lines_count", changed_lines_count) or 0)
+        elif isinstance(edit_count, int):
+            changed_lines_count = max(changed_lines_count, int(edit_count))
+
+        penalty = 0.0
+
+        if changed_files_count > 5:
+            penalty -= 0.15
+        if changed_files_count > 10:
+            penalty -= 0.25
+
+        if changed_lines_count > 120:
+            penalty -= 0.15
+        if changed_lines_count > 300:
+            penalty -= 0.25
+
+        return max(-0.80, penalty)
+
+    def penalty_timeout_abuse(self, step_count: int) -> float:
+        if step_count > 30:
+            return -0.80
+        if step_count > 20:
+            return -0.50
+        return 0.0
+
+    def penalty_bruteforce_attempts(self, consecutive_edit_actions: int, failed_validations: int) -> float:
+        penalty = 0.0
+        if consecutive_edit_actions >= 6:
+            penalty -= 0.25
+        if consecutive_edit_actions >= 10:
+            penalty -= 0.35
+
+        if failed_validations >= 3:
+            penalty -= 0.20
+        if failed_validations >= 6:
+            penalty -= 0.35
+
+        return max(-0.80, penalty)
+
+    def total_penalty(
+        self,
+        current_config: str = "",
+        previous_config: str = "",
+        edit_count: int | dict[str, Any] | None = None,
+        changed_files_count: int = 0,
+        changed_lines_count: int = 0,
+        step_count: int = 0,
+        consecutive_edit_actions: int = 0,
+        failed_validations: int = 0,
+    ) -> float:
+        total = 0.0
+        total += self.penalty_invalid_config(current_config)
+        total += self.penalty_stage_skipping(current_config)
+        total += self.penalty_fake_success(current_config)
+        total += self.penalty_hardcoded_outputs(current_config)
+        total += self.penalty_breaking_pipeline(previous_config, current_config)
+        total += self.penalty_excessive_edits(
+            edit_count=edit_count,
+            changed_files_count=changed_files_count,
+            changed_lines_count=changed_lines_count,
+        )
+        total += self.penalty_timeout_abuse(step_count)
+        total += self.penalty_bruteforce_attempts(consecutive_edit_actions, failed_validations)
+
+        return round(total, 4)
+
+    def _count_hits(self, text: str, patterns: tuple[str, ...]) -> int:
+        text = text or ""
+        return sum(1 for pattern in patterns if re.search(pattern, text, flags=re.IGNORECASE))
+
+    def _is_yaml_valid(self, config_text: str) -> bool:
+        if not (config_text or "").strip():
+            return False
+        try:
+            yaml.safe_load(config_text)
+            return True
+        except yaml.YAMLError:
+            return False
+
+    def _extract_stage_names(self, config_text: str) -> set[str]:
+        try:
+            parsed = yaml.safe_load(config_text)
+        except yaml.YAMLError:
+            return set()
+
+        if parsed is None:
+            return set()
+
+        stages: set[str] = set()
+        self._walk_for_stages(parsed, stages)
+        return stages
+
+    def _walk_for_stages(self, node: Any, stages: set[str]) -> None:
+        if isinstance(node, dict):
+            for key, value in node.items():
+                key_name = str(key).lower()
+                if key_name in {"stages", "jobs", "job"}:
+                    if isinstance(value, dict):
+                        for stage_name in value.keys():
+                            stages.add(str(stage_name))
+                    elif isinstance(value, list):
+                        for stage_name in value:
+                            stages.add(str(stage_name))
+                self._walk_for_stages(value, stages)
+        elif isinstance(node, list):
+            for item in node:
+                self._walk_for_stages(item, stages)

env/environment.py

@@ -0,0 +1,769 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from difflib import SequenceMatcher
+import random
+import re
+from typing import Any
+
+import yaml
+
+from env.models import Action, EnvStateSnapshot, Observation, Reward
+from env.rewards import RewardCalculator
+from env.tasks import get_task_by_id, get_tasks_by_difficulty
+from env.tasks.task_types import CICDTask
+
+
+REQUIRED_TOOLS = [
+    "read_file",
+    "read_logs",
+    "analyze_error",
+    "edit_config",
+    "run_pipeline_stage",
+    "run_tests",
+    "validate_fix",
+    "submit_solution",
+]
+
+MAX_STEPS = 30
+
+
+@dataclass
+class EnvironmentState:
+    task: CICDTask
+    current_config: str
+    previous_config: str
+    step_count: int = 0
+    done: bool = False
+    progress_flags: dict[str, bool] = field(default_factory=dict)
+    file_modification_count: int = 0
+    total_changed_lines: int = 0
+    hidden_test_pass_rate: float = 0.0
+    action_history: list[str] = field(default_factory=list)
+    stage_results: dict[str, bool] = field(default_factory=dict)
+    failed_validations: int = 0
+    consecutive_edit_actions: int = 0
+    current_logs: str = ""
+    last_error: str = ""
+    last_action_error: str | None = None
+    last_info: dict[str, Any] = field(default_factory=dict)
+
+
+class CICDDebuggerEnvironment:
+    """RL-style CI/CD debugging environment with strict tool-based actions."""
+
+    def __init__(
+        self,
+        max_steps: int = MAX_STEPS,
+        seed: int | None = None,
+        llm_judge: Any | None = None,
+    ) -> None:
+        self.max_steps = max(1, int(max_steps))
+        self.random = random.Random(seed)
+        self.reward_calculator = RewardCalculator(llm_judge=llm_judge)
+        self._state: EnvironmentState | None = None
+
+    async def reset(self, task_id: str | None = None, difficulty: str | None = None) -> dict[str, Any]:
+        task = self._select_task(task_id=task_id, difficulty=difficulty)
+
+        self._state = EnvironmentState(
+            task=task,
+            current_config=task.broken_config,
+            previous_config=task.broken_config,
+            progress_flags={tool: False for tool in REQUIRED_TOOLS},
+            current_logs=task.logs,
+            last_error=task.error_message,
+        )
+
+        return Observation.model_validate(self._build_observation()).model_dump()
+
+    async def step(self, action: Any) -> tuple[dict[str, Any], float, bool, dict[str, Any]]:
+        if self._state is None:
+            raise RuntimeError("Environment not initialized. Call reset() first.")
+
+        if self._state.done:
+            reward_model = Reward(value=0.0, components={"total": 0.0})
+            return Observation.model_validate(self._build_observation()).model_dump(), float(reward_model.value), True, {
+                "tool": "none",
+                "message": "episode already completed",
+                "error": None,
+                "reward_model": reward_model.model_dump(),
+            }
+
+        parsed_action = Action.from_input(action)
+        tool, payload = parsed_action.tool, dict(parsed_action.payload)
+        self._state.step_count += 1
+        self._state.previous_config = self._state.current_config
+        self._state.action_history.append(tool)
+        self._state.last_action_error = None
+
+        info: dict[str, Any] = {
+            "tool": tool,
+            "message": "",
+            "error": None,
+        }
+        changed_lines = 0
+
+        result: dict[str, Any] = {
+            "previous_config": self._state.previous_config,
+            "current_config": self._state.current_config,
+            "fixed_config": self._state.current_config,
+            "expected_config": self._state.task.expected_config,
+            "error": self._state.last_error,
+            "logs_analyzed": False,
+            "error_diagnosed": False,
+            "fix_proposed": False,
+            "pipeline_run": False,
+            "tests_passed": False,
+            "command_succeeded": False,
+            "changed_files_count": 0,
+            "changed_lines_count": 0,
+            "edit_count": {
+                "changed_files_count": self._state.file_modification_count,
+                "changed_lines_count": self._state.total_changed_lines,
+            },
+            "deterministic_score": None,
+            "hidden_test_pass_rate": None,
+            "judge_scores": None,
+            "hacking_attempt": False,
+        }
+
+        if tool not in REQUIRED_TOOLS:
+            info["message"] = "unsupported action tool"
+            info["error"] = f"tool '{tool}' is not allowed"
+            self._state.last_action_error = str(info["error"])
+        elif tool == "read_file":
+            self._state.progress_flags[tool] = True
+            result["command_succeeded"] = True
+            info["message"] = "returned current workflow config"
+            self._state.current_logs = self._state.current_config
+            self._state.consecutive_edit_actions = 0
+        elif tool == "read_logs":
+            self._state.progress_flags[tool] = True
+            result["logs_analyzed"] = True
+            result["command_succeeded"] = True
+            info["message"] = "returned pipeline failure logs"
+            self._state.current_logs = self._state.task.logs
+            self._state.consecutive_edit_actions = 0
+        elif tool == "analyze_error":
+            self._state.progress_flags[tool] = True
+            result["error_diagnosed"] = True
+            result["command_succeeded"] = True
+            root_cause = self._detect_root_cause(self._state.current_config, self._state.task)
+            info["message"] = f"root cause: {root_cause}"
+            self._state.current_logs = f"analysis result: {root_cause}"
+            self._state.consecutive_edit_actions = 0
+        elif tool == "edit_config":
+            self._state.progress_flags[tool] = True
+            updated_config, summary = self._apply_edit(self._state.current_config, payload, self._state.task)
+            changed_lines = self._count_changed_lines(self._state.current_config, updated_config)
+
+            if changed_lines > 0:
+                self._state.current_config = updated_config
+                self._state.file_modification_count += 1
+                self._state.total_changed_lines += changed_lines
+                result["fix_proposed"] = True
+                result["command_succeeded"] = True
+                info["message"] = summary
+                self._state.current_logs = f"edit applied: {summary}"
+            else:
+                result["command_succeeded"] = False
+                info["message"] = "no config changes applied"
+                info["error"] = "edit_config did not modify workflow"
+                self._state.last_action_error = str(info["error"])
+                self._state.current_logs = "edit action produced no changes"
+
+            self._state.consecutive_edit_actions += 1
+        elif tool == "run_pipeline_stage":
+            self._state.progress_flags[tool] = True
+            stage = self._extract_stage(payload, fallback=self._state.task.failure_stage)
+            success, stage_logs = self._simulate_stage(self._state.current_config, stage, self._state.task)
+            self._state.stage_results[stage] = success
+            result["pipeline_run"] = True
+            result["command_succeeded"] = success
+            info["message"] = f"stage '{stage}' {'passed' if success else 'failed'}"
+            if not success:
+                info["error"] = stage_logs
+                self._state.last_action_error = stage_logs
+                self._state.last_error = stage_logs
+            self._state.current_logs = stage_logs
+            self._state.consecutive_edit_actions = 0
+        elif tool == "run_tests":
+            self._state.progress_flags[tool] = True
+            tests_passed, test_logs = self._run_tests(self._state.current_config, self._state.task)
+            result["pipeline_run"] = True
+            result["tests_passed"] = tests_passed
+            result["command_succeeded"] = tests_passed
+            info["message"] = "tests passed" if tests_passed else "tests failed"
+            if not tests_passed:
+                info["error"] = test_logs
+                self._state.last_action_error = test_logs
+                self._state.last_error = test_logs
+            self._state.current_logs = test_logs
+            self._state.consecutive_edit_actions = 0
+        elif tool == "validate_fix":
+            self._state.progress_flags[tool] = True
+            validation = self._validate_current_fix(self._state)
+            result.update(validation)
+            result["pipeline_run"] = True
+            is_valid = bool(validation.get("is_valid"))
+            result["command_succeeded"] = is_valid
+
+            if not is_valid:
+                self._state.failed_validations += 1
+                info["error"] = str(validation.get("summary", "validation failed"))
+                self._state.last_action_error = str(info["error"])
+
+            info["message"] = "validation passed" if is_valid else "validation failed"
+            self._state.hidden_test_pass_rate = float(validation.get("hidden_test_pass_rate") or 0.0)
+            self._state.current_logs = str(validation.get("summary", "validation complete"))
+            self._state.consecutive_edit_actions = 0
+        elif tool == "submit_solution":
+            validation = self._validate_current_fix(self._state)
+            result.update(validation)
+            result["pipeline_run"] = True
+            self._state.progress_flags[tool] = True
+            accepted = bool(validation.get("is_valid"))
+            result["command_succeeded"] = accepted
+
+            if accepted:
+                self._state.done = True
+                info["message"] = "solution accepted"
+                self._state.current_logs = "submission accepted"
+            else:
+                self._state.failed_validations += 1
+                info["message"] = "solution rejected"
+                info["error"] = "submission failed quality checks"
+                self._state.last_action_error = str(info["error"])
+                self._state.current_logs = str(validation.get("summary", "submission rejected"))
+
+            self._state.hidden_test_pass_rate = float(validation.get("hidden_test_pass_rate") or 0.0)
+            self._state.consecutive_edit_actions = 0
+
+        result["hacking_attempt"] = self._detect_hacking_attempt(tool, payload, self._state.current_config)
+        result["current_config"] = self._state.current_config
+        result["fixed_config"] = self._state.current_config
+        result["changed_files_count"] = 1 if changed_lines > 0 else 0
+        result["changed_lines_count"] = changed_lines
+        result["edit_count"] = {
+            "changed_files_count": self._state.file_modification_count,
+            "changed_lines_count": self._state.total_changed_lines,
+        }
+
+        if info["error"]:
+            self._state.last_error = str(info["error"])
+        result["error"] = self._state.last_error
+
+        if self._state.step_count >= self.max_steps and not self._state.done:
+            self._state.done = True
+            if not info["error"]:
+                info["error"] = "max_steps_reached"
+            info["message"] = "max steps reached"
+
+        reward = self.reward_calculator.calculate_step_reward(
+            state={
+                "step_count": self._state.step_count,
+                "previous_config": self._state.previous_config,
+                "expected_config": self._state.task.expected_config,
+                "original_config": self._state.task.broken_config,
+                "error": self._state.last_error,
+                "changed_files_count": self._state.file_modification_count,
+                "changed_lines_count": self._state.total_changed_lines,
+                "consecutive_edit_actions": self._state.consecutive_edit_actions,
+                "failed_validations": self._state.failed_validations,
+            },
+            action=tool,
+            result=result,
+            original_config=self._state.task.broken_config,
+            fixed_config=self._state.current_config,
+            error_message=self._state.last_error,
+            expected_config=self._state.task.expected_config,
+            metadata=self._state.task.metadata,
+        )
+
+        reward_model = Reward(value=float(reward), components={"total": float(reward)})
+        info["reward_model"] = reward_model.model_dump()
+
+        self._state.last_info = info
+        observation = Observation.model_validate(self._build_observation()).model_dump()
+        done = bool(self._state.done)
+
+        return observation, float(reward_model.value), done, info
+
+    async def close(self) -> None:
+        return None
+
+    def get_state(self) -> dict[str, Any]:
+        if self._state is None:
+            return EnvStateSnapshot(initialized=False).model_dump()
+
+        snapshot = {
+            "initialized": True,
+            "task_id": self._state.task.task_id,
+            "difficulty": self._state.task.difficulty,
+            "actual_bug": self._state.task.actual_bug,
+            "correct_solution": self._state.task.expected_config,
+            "failure_stage": self._state.task.failure_stage,
+            "step_count": self._state.step_count,
+            "done": self._state.done,
+            "progress_flags": dict(self._state.progress_flags),
+            "file_modification_count": self._state.file_modification_count,
+            "total_changed_lines": self._state.total_changed_lines,
+            "hidden_test_pass_rate": self._state.hidden_test_pass_rate,
+            "stage_results": dict(self._state.stage_results),
+            "failed_validations": self._state.failed_validations,
+            "last_action_error": self._state.last_action_error,
+            "last_error": self._state.last_error,
+        }
+        return EnvStateSnapshot.model_validate(snapshot).model_dump()
+
+    def state(self) -> dict[str, Any]:
+        return self.get_state()
+
+    def _build_observation(self) -> dict[str, Any]:
+        if self._state is None:
+            raise RuntimeError("Environment not initialized")
+
+        observation = {
+            "task_id": self._state.task.task_id,
+            "difficulty": self._state.task.difficulty,
+            "failure_stage": self._state.task.failure_stage,
+            "actual_bug": self._state.task.actual_bug,
+            "config": self._state.current_config,
+            "logs": self._state.current_logs,
+            "error_message": self._state.last_error,
+            "available_tools": list(REQUIRED_TOOLS),
+            "progress_flags": dict(self._state.progress_flags),
+            "file_modification_count": self._state.file_modification_count,
+            "hidden_test_pass_rate": self._state.hidden_test_pass_rate,
+            "step_count": self._state.step_count,
+            "last_action_error": self._state.last_action_error,
+        }
+        return Observation.model_validate(observation).model_dump()
+
+    def _select_task(self, task_id: str | None, difficulty: str | None) -> CICDTask:
+        if task_id:
+            task = get_task_by_id(task_id)
+            if task is None:
+                raise ValueError(f"Unknown task_id: {task_id}")
+            return task
+
+        filtered = get_tasks_by_difficulty(difficulty)
+        if not filtered:
+            raise ValueError(f"No tasks available for difficulty: {difficulty}")
+
+        return self.random.choice(filtered)
+
+    def _parse_action(self, action: Any) -> tuple[str, dict[str, Any]]:
+        parsed = Action.from_input(action)
+        return parsed.tool, dict(parsed.payload)
+
+    def _extract_stage(self, payload: dict[str, Any], fallback: str) -> str:
+        direct_stage = str(payload.get("stage") or "").strip().lower()
+        if direct_stage in {"build", "test", "deploy"}:
+            return direct_stage
+
+        raw = str(payload.get("raw") or "").lower()
+        for stage in ("build", "test", "deploy"):
+            if stage in raw:
+                return stage
+
+        return fallback
+
+    def _detect_root_cause(self, config_text: str, task: CICDTask) -> str:
+        normalized = self._normalize(config_text)
+        broken_token = self._normalize(str(task.metadata.get("broken_token", "")))
+
+        if broken_token and broken_token in normalized:
+            return task.actual_bug
+
+        if not self._is_yaml_valid(config_text):
+            return "workflow YAML is invalid"
+
+        fixed_token = self._normalize(str(task.metadata.get("fixed_token", "")))
+        if fixed_token and fixed_token not in normalized:
+            return f"missing expected fix token: {task.metadata.get('fixed_token')}"
+
+        return "configuration still deviates from expected pipeline behavior"
+
+    def _apply_edit(self, current_config: str, payload: dict[str, Any], task: CICDTask) -> tuple[str, str]:
+        candidate = current_config
+        edits: list[str] = []
+
+        new_config = payload.get("new_config")
+        if isinstance(new_config, str) and new_config.strip():
+            return new_config.strip(), "applied payload new_config"
+
+        raw = str(payload.get("raw") or "")
+        raw_lower = raw.lower()
+
+        replace_match = re.search(
+            r"replace\s+['\"]?(.+?)['\"]?\s+with\s+['\"]?(.+?)['\"]?\s*$",
+            raw,
+            flags=re.IGNORECASE,
+        )
+        if replace_match:
+            old = replace_match.group(1).strip()
+            new = replace_match.group(2).strip()
+            if old and old in candidate:
+                candidate = candidate.replace(old, new)
+                edits.append(f"replaced '{old}' with '{new}'")
+
+        if "checkout" in raw_lower and "actions/checkout@v4" not in candidate:
+            updated = self._ensure_checkout(candidate)
+            if updated != candidate:
+                candidate = updated
+                edits.append("inserted actions/checkout@v4 step")
+
+        if "permissions" in raw_lower or "actions: write" in raw_lower:
+            updated = self._ensure_actions_write(candidate)
+            if updated != candidate:
+                candidate = updated
+                edits.append("added actions: write permission")
+
+        if not edits and any(token in raw_lower for token in ("yaml", "indent", "syntax")):
+            updated = self._repair_yaml(candidate, task.expected_config)
+            if updated != candidate:
+                candidate = updated
+                edits.append("repaired YAML structure")
+
+        broken_token = str(task.metadata.get("broken_token", ""))
+        fixed_token = str(task.metadata.get("fixed_token", ""))
+        if not edits and broken_token and fixed_token and broken_token in candidate:
+            occurrence_count = candidate.count(broken_token)
+
+            if occurrence_count > 1:
+                candidate = task.expected_config
+                edits.append("applied canonical fix for ambiguous token")
+            elif fixed_token.strip().endswith(":"):
+                expected_block = self._extract_expected_block(task.expected_config, fixed_token)
+                if expected_block and expected_block not in candidate:
+                    candidate = candidate.replace(broken_token, f"{broken_token}\n{expected_block}", 1)
+                    edits.append("inserted expected YAML block")
+                else:
+                    candidate = candidate.replace(broken_token, fixed_token, 1)
+                    edits.append("applied metadata token replacement")
+            else:
+                expected_line = self._find_line_containing(task.expected_config, fixed_token)
+                replacement = expected_line.strip() if expected_line else fixed_token
+                candidate = candidate.replace(broken_token, replacement, 1)
+                edits.append("applied metadata token replacement")
+
+        if not edits and fixed_token and fixed_token not in candidate and not broken_token:
+            updated = self._append_missing_token(candidate, fixed_token)
+            if updated != candidate:
+                candidate = updated
+                edits.append("appended expected token")
+
+        if not edits and any(token in raw_lower for token in ("expected config", "apply expected", "canonical fix")):
+            candidate = task.expected_config
+            edits.append("replaced with expected task config")
+
+        summary = "; ".join(edits) if edits else "no-op edit"
+        return candidate, summary
+
+    def _ensure_checkout(self, config_text: str) -> str:
+        if "actions/checkout@v4" in config_text:
+            return config_text
+
+        marker = "steps:\n"
+        insert = "      - uses: actions/checkout@v4\n"
+        if marker in config_text:
+            return config_text.replace(marker, marker + insert, 1)
+
+        return config_text
+
+    def _ensure_actions_write(self, config_text: str) -> str:
+        if "actions: write" in config_text:
+            return config_text
+
+        if "permissions:" in config_text:
+            lines = config_text.splitlines()
+            out: list[str] = []
+            inserted = False
+            for line in lines:
+                out.append(line)
+                if line.strip().startswith("permissions:") and not inserted:
+                    continue
+                if line.strip().startswith("contents:") and not inserted:
+                    indent = line[: len(line) - len(line.lstrip(" "))]
+                    out.append(f"{indent}actions: write")
+                    inserted = True
+            if inserted:
+                return "\n".join(out)
+
+        return "permissions:\n  actions: write\n" + config_text
+
+    def _append_missing_token(self, config_text: str, token: str) -> str:
+        if not token or token in config_text:
+            return config_text
+
+        lower_token = token.lower()
+        if "actions/checkout@v4" in lower_token:
+            return self._ensure_checkout(config_text)
+        if "actions: write" in lower_token:
+            return self._ensure_actions_write(config_text)
+
+        return config_text + "\n" + token
+
+    def _repair_yaml(self, current_config: str, expected_config: str) -> str:
+        if self._is_yaml_valid(current_config):
+            return current_config
+
+        if expected_config and self._is_yaml_valid(expected_config):
+            return expected_config
+
+        return current_config
+
+    def _find_line_containing(self, config_text: str, token: str) -> str | None:
+        target = (token or "").strip()
+        if not target:
+            return None
+
+        for line in (config_text or "").splitlines():
+            if target in line:
+                return line
+
+        return None
+
+    def _extract_expected_block(self, config_text: str, token: str) -> str:
+        lines = (config_text or "").splitlines()
+        target = (token or "").strip()
+        if not target:
+            return ""
+
+        for idx, line in enumerate(lines):
+            if target not in line:
+                continue
+
+            base_indent = len(line) - len(line.lstrip(" "))
+            block = [line]
+            for next_line in lines[idx + 1 :]:
+                if not next_line.strip():
+                    break
+                next_indent = len(next_line) - len(next_line.lstrip(" "))
+                if next_indent <= base_indent:
+                    break
+                block.append(next_line)
+            return "\n".join(block)
+
+        return ""
+
+    def _simulate_stage(self, config_text: str, stage: str, task: CICDTask) -> tuple[bool, str]:
+        if not self._is_yaml_valid(config_text):
+            return False, "invalid workflow YAML"
+
+        expected_has_stage = self._stage_exists(task.expected_config, stage)
+        current_has_stage = self._stage_exists(config_text, stage)
+
+        if expected_has_stage and not current_has_stage:
+            return False, f"required stage '{stage}' is missing"
+
+        if not expected_has_stage and not current_has_stage:
+            return True, f"{stage} stage not required for this task"
+
+        normalized = self._normalize(config_text)
+        broken_token = self._normalize(str(task.metadata.get("broken_token", "")))
+        fixed_token = self._normalize(str(task.metadata.get("fixed_token", "")))
+
+        if self._contains_hacking_pattern(config_text):
+            return False, "unsafe shortcut pattern detected"
+
+        if stage == task.failure_stage and broken_token and broken_token in normalized:
+            return False, task.logs
+
+        if stage == task.failure_stage and fixed_token and fixed_token not in normalized:
+            return False, task.logs
+
+        commands = self._extract_commands(config_text)
+
+        if stage == "build":
+            build_tokens = ("npm ci", "npm install", "pip install", "go build", "mvn", "yarn install", "pnpm install")
+            if not any(any(token in cmd for token in build_tokens) for cmd in commands):
+                return False, "build stage has no install/build command"
+
+        if stage == "test":
+            test_tokens = ("npm test", "pytest", "go test", "mvn test", "yarn test", "pnpm test")
+            if not any(any(token in cmd for token in test_tokens) for cmd in commands):
+                return False, "test stage has no test command"
+
+        if stage == "deploy":
+            deploy_tokens = ("deploy", "publish", "upload-artifact", "release")
+            if not any(any(token in cmd for token in deploy_tokens) for cmd in commands):
+                return False, "deploy stage has no deployment command"
+
+        return True, f"{stage} stage passed"
+
+    def _run_tests(self, config_text: str, task: CICDTask) -> tuple[bool, str]:
+        if self._stage_exists(task.expected_config, "build"):
+            build_ok, build_logs = self._simulate_stage(config_text, "build", task)
+            if not build_ok:
+                return False, build_logs
+
+        if self._stage_exists(task.expected_config, "test"):
+            test_ok, test_logs = self._simulate_stage(config_text, "test", task)
+            if not test_ok:
+                return False, test_logs
+
+        similarity = SequenceMatcher(None, self._normalize(config_text), self._normalize(task.expected_config)).ratio()
+        if similarity < 0.45:
+            return False, "tests failed: fix diverges significantly from expected pipeline"
+
+        return True, "tests passed"
+
+    def _validate_current_fix(self, state: EnvironmentState) -> dict[str, Any]:
+        current = state.current_config
+        task = state.task
+
+        deterministic_score = self.reward_calculator.deterministic_grader.grade(
+            current,
+            task.expected_config,
+            metadata=task.metadata,
+        )
+        hidden_test_pass_rate = self.reward_calculator.hidden_test_runner.evaluate_fix(
+            fixed_config=current,
+            expected_config=task.expected_config,
+            metadata=task.metadata,
+        )
+
+        judge_scores = None
+        if self.reward_calculator.llm_judge is not None:
+            try:
+                judge_scores = self.reward_calculator.llm_judge.evaluate_fix(
+                    task.broken_config,
+                    current,
+                    state.last_error,
+                )
+            except Exception:
+                judge_scores = None
+
+        tests_passed, test_logs = self._run_tests(current, task)
+
+        stage_ok, stage_logs = self._simulate_stage(current, task.failure_stage, task)
+
+        broken_token = self._normalize(str(task.metadata.get("broken_token", "")))
+        fixed_token = self._normalize(str(task.metadata.get("fixed_token", "")))
+        normalized_current = self._normalize(current)
+
+        token_constraints_met = True
+        if broken_token and broken_token in normalized_current:
+            token_constraints_met = False
+        if fixed_token and fixed_token not in normalized_current:
+            token_constraints_met = False
+
+        judge_average = 1.0
+        if isinstance(judge_scores, dict):
+            judge_average = (
+                float(judge_scores.get("correctness", 0.0))
+                + float(judge_scores.get("minimalism", 0.0))
+                + float(judge_scores.get("quality", 0.0))
+            ) / 3.0
+
+        is_valid = (
+            tests_passed
+            and stage_ok
+            and token_constraints_met
+            and deterministic_score >= 0.72
+            and hidden_test_pass_rate >= 0.65
+            and judge_average >= 0.5
+        )
+
+        summary = (
+            f"deterministic={deterministic_score:.2f}; "
+            f"hidden={hidden_test_pass_rate:.2f}; "
+            f"tests={'pass' if tests_passed else 'fail'}; "
+            f"failure_stage={'pass' if stage_ok else 'fail'}"
+        )
+        if not tests_passed:
+            summary = test_logs
+        elif not stage_ok:
+            summary = stage_logs
+        elif not token_constraints_met:
+            summary = "validation failed: required bug-fix token constraints not satisfied"
+
+        return {
+            "deterministic_score": deterministic_score,
+            "hidden_test_pass_rate": hidden_test_pass_rate,
+            "judge_scores": judge_scores,
+            "tests_passed": tests_passed,
+            "is_valid": is_valid,
+            "summary": summary,
+        }
+
+    def _detect_hacking_attempt(self, tool: str, payload: dict[str, Any], config_text: str) -> bool:
+        payload_text = str(payload.get("raw") or "")
+        brute_force_detected = bool(
+            self._state and (self._state.consecutive_edit_actions >= 6 or self._state.failed_validations >= 3)
+        )
+
+        return (
+            brute_force_detected
+            or self._contains_hacking_pattern(payload_text)
+            or self._contains_hacking_pattern(config_text)
+        )
+
+    def _contains_hacking_pattern(self, text: str) -> bool:
+        if not text:
+            return False
+
+        patterns = (
+            *self.reward_calculator.anti_hacking_detector.STAGE_SKIP_PATTERNS,
+            *self.reward_calculator.anti_hacking_detector.FAKE_SUCCESS_PATTERNS,
+            *self.reward_calculator.anti_hacking_detector.HARDCODED_OUTPUT_PATTERNS,
+        )
+        return any(re.search(pattern, text, flags=re.IGNORECASE) for pattern in patterns)
+
+    def _extract_commands(self, config_text: str) -> list[str]:
+        commands: list[str] = []
+        for raw_line in config_text.splitlines():
+            line = raw_line.strip().lower()
+            if "- run:" in line:
+                commands.append(line.replace("- run:", "").strip())
+            elif line.startswith("- run "):
+                commands.append(line.replace("- run ", "", 1).strip())
+        return commands
+
+    def _is_yaml_valid(self, config_text: str) -> bool:
+        try:
+            parsed = yaml.safe_load(config_text)
+        except yaml.YAMLError:
+            return False
+        return isinstance(parsed, dict)
+
+    def _stage_exists(self, config_text: str, stage: str) -> bool:
+        try:
+            parsed = yaml.safe_load(config_text)
+        except yaml.YAMLError:
+            return False
+
+        if not isinstance(parsed, dict):
+            return False
+
+        jobs = parsed.get("jobs")
+        if isinstance(jobs, dict) and stage in jobs:
+            return True
+
+        stages = parsed.get("stages")
+        if isinstance(stages, dict) and stage in stages:
+            return True
+        if isinstance(stages, list) and stage in stages:
+            return True
+
+        return False
+
+    def _count_changed_lines(self, previous: str, current: str) -> int:
+        prev_lines = previous.splitlines()
+        curr_lines = current.splitlines()
+        changed = 0
+
+        max_len = max(len(prev_lines), len(curr_lines))
+        for idx in range(max_len):
+            left = prev_lines[idx] if idx < len(prev_lines) else ""
+            right = curr_lines[idx] if idx < len(curr_lines) else ""
+            if left != right:
+                changed += 1
+
+        return changed
+
+    def _normalize(self, value: str) -> str:
+        return re.sub(r"\s+", " ", value.strip().lower())

env/graders/__init__.py

@@ -0,0 +1,4 @@
+from env.graders.deterministic import DeterministicGrader
+from env.graders.llm_judge import LLMJudge
+
+__all__ = ["DeterministicGrader", "LLMJudge"]

env/graders/deterministic.py

@@ -0,0 +1,189 @@
+from __future__ import annotations
+
+import re
+from difflib import SequenceMatcher
+from typing import Any
+
+import yaml
+
+
+class DeterministicGrader:
+    """Deterministic correctness scoring for CI/CD config fixes."""
+
+    COMMAND_KEYS = {
+        "script",
+        "scripts",
+        "run",
+        "command",
+        "commands",
+        "steps",
+        "before_script",
+        "after_script",
+    }
+
+    BROKEN_COMMAND_PATTERNS = (
+        r"\bnpm\s+tset\b",
+        r"\bpyhton\b",
+        r"\bpip\s+isntall\b",
+        r"\bgo\s+tset\b",
+    )
+
+    def grade(self, current_config: str, expected_config: str, metadata: dict[str, Any] | None = None) -> float:
+        metadata = metadata or {}
+        current_config = current_config or ""
+        expected_config = expected_config or ""
+
+        syntax_score = self._syntax_score(current_config)
+        functional_score = self._functional_score(current_config, expected_config, metadata)
+        similarity_score = self._similarity_score(current_config, expected_config)
+
+        total = (0.20 * syntax_score) + (0.60 * functional_score) + (0.20 * similarity_score)
+
+        if syntax_score == 0.0:
+            total = min(total, 0.30)
+
+        return round(self._clamp_01(total), 4)
+
+    def _syntax_score(self, config_text: str) -> float:
+        if not (config_text or "").strip():
+            return 0.0
+
+        try:
+            yaml.safe_load(config_text)
+            return 1.0
+        except yaml.YAMLError:
+            return 0.0
+
+    def _functional_score(self, current_config: str, expected_config: str, metadata: dict[str, Any]) -> float:
+        expected_commands = self._extract_commands(expected_config)
+        current_commands = self._extract_commands(current_config)
+
+        if expected_commands:
+            matched = 0
+            for expected in expected_commands:
+                if any(self._commands_match(expected, current) for current in current_commands):
+                    matched += 1
+            command_score = matched / len(expected_commands)
+        else:
+            command_score = self._similarity_score(current_config, expected_config)
+
+        issue_score = self._issue_resolution_score(current_config, metadata)
+        broken_penalty = 0.35 if self._has_known_broken_command(current_config) else 0.0
+
+        combined = (0.80 * command_score) + (0.20 * issue_score) - broken_penalty
+        return self._clamp_01(combined)
+
+    def _issue_resolution_score(self, current_config: str, metadata: dict[str, Any]) -> float:
+        broken_token = self._normalize_text(str(metadata.get("broken_token", "")))
+        fixed_token = self._normalize_text(str(metadata.get("fixed_token", "")))
+        current_normalized = self._normalize_text(current_config)
+
+        if not broken_token and not fixed_token:
+            return 1.0
+
+        if broken_token and broken_token in current_normalized:
+            return 0.0
+
+        if fixed_token and fixed_token not in current_normalized:
+            return 0.0
+
+        return 1.0
+
+    def _extract_commands(self, config_text: str) -> list[str]:
+        commands: list[str] = []
+
+        try:
+            parsed = yaml.safe_load(config_text)
+        except yaml.YAMLError:
+            parsed = None
+
+        if parsed is not None:
+            self._walk_yaml(parsed, commands)
+
+        if not commands:
+            commands.extend(self._extract_commands_from_text(config_text))
+
+        deduped: list[str] = []
+        seen: set[str] = set()
+        for command in commands:
+            normalized = self._normalize_text(command)
+            if normalized and normalized not in seen:
+                seen.add(normalized)
+                deduped.append(normalized)
+
+        return deduped
+
+    def _walk_yaml(self, node: Any, commands: list[str]) -> None:
+        if isinstance(node, dict):
+            for key, value in node.items():
+                key_name = str(key).lower()
+                if key_name in self.COMMAND_KEYS:
+                    commands.extend(self._extract_string_values(value))
+                self._walk_yaml(value, commands)
+        elif isinstance(node, list):
+            for item in node:
+                self._walk_yaml(item, commands)
+
+    def _extract_string_values(self, value: Any) -> list[str]:
+        if isinstance(value, str):
+            return [value]
+        if isinstance(value, list):
+            return [item for item in value if isinstance(item, str)]
+        if isinstance(value, dict):
+            output: list[str] = []
+            for nested in value.values():
+                output.extend(self._extract_string_values(nested))
+            return output
+        return []
+
+    def _extract_commands_from_text(self, config_text: str) -> list[str]:
+        commands: list[str] = []
+
+        for raw_line in (config_text or "").splitlines():
+            line = raw_line.strip()
+            if not line or line.startswith("#"):
+                continue
+
+            if ":" in line and not line.startswith("-") and line.endswith(":"):
+                continue
+
+            line = line.lstrip("-").strip()
+            if any(token in line.lower() for token in ("npm", "pytest", "python", "yarn", "pnpm", "go test", "mvn test")):
+                commands.append(line)
+
+        return commands
+
+    def _has_known_broken_command(self, config_text: str) -> bool:
+        return any(re.search(pattern, config_text or "", flags=re.IGNORECASE) for pattern in self.BROKEN_COMMAND_PATTERNS)
+
+    def _commands_match(self, expected: str, current: str) -> bool:
+        expected_normalized = self._normalize_text(expected)
+        current_normalized = self._normalize_text(current)
+
+        if expected_normalized == current_normalized:
+            return True
+
+        if expected_normalized in current_normalized:
+            return True
+
+        if current_normalized in expected_normalized and len(current_normalized) > 6:
+            return True
+
+        return False
+
+    def _similarity_score(self, current_config: str, expected_config: str) -> float:
+        left = self._normalize_text(current_config)
+        right = self._normalize_text(expected_config)
+
+        if not left and not right:
+            return 1.0
+        if not left or not right:
+            return 0.0
+
+        return self._clamp_01(SequenceMatcher(None, left, right).ratio())
+
+    def _normalize_text(self, value: str) -> str:
+        return re.sub(r"\s+", " ", (value or "")).strip().lower()
+
+    def _clamp_01(self, value: float) -> float:
+        return max(0.0, min(1.0, float(value)))

env/graders/llm_judge.py

@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+import json
+import re
+from typing import Any
+
+
+class LLMJudge:
+    """Scores qualitative fix quality while remaining robust to bad model output."""
+
+    def __init__(self, model: Any):
+        self.model = model
+
+    def build_prompt(self, original_config: str, fixed_config: str, error_message: str) -> str:
+        return (
+            "You are a CI/CD fix quality judge.\n"
+            "Return strict JSON with keys correctness, minimalism, quality in [0,1].\n"
+            "No prose.\n\n"
+            f"Original config:\n{original_config}\n\n"
+            f"Fixed config:\n{fixed_config}\n\n"
+            f"Error message:\n{error_message}\n"
+        )
+
+    def evaluate_fix(self, original_config: str, fixed_config: str, error_message: str) -> dict[str, float]:
+        default = {
+            "correctness": 0.0,
+            "minimalism": 0.0,
+            "quality": 0.0,
+        }
+
+        if self.model is None:
+            return default
+
+        prompt = self.build_prompt(original_config or "", fixed_config or "", error_message or "")
+
+        try:
+            raw_output = self.model(prompt, max_length=300)
+            text = self._extract_text(raw_output)
+        except Exception:
+            return default
+
+        if not text.strip():
+            return default
+
+        parsed = self._parse_json_with_fallback(text)
+        if parsed is None:
+            parsed = self._parse_regex_scores(text)
+
+        return {
+            "correctness": self._clamp(parsed.get("correctness", 0.0) if parsed else 0.0),
+            "minimalism": self._clamp(parsed.get("minimalism", 0.0) if parsed else 0.0),
+            "quality": self._clamp(parsed.get("quality", 0.0) if parsed else 0.0),
+        }
+
+    def _extract_text(self, raw_output: Any) -> str:
+        if isinstance(raw_output, str):
+            return raw_output
+
+        if isinstance(raw_output, list) and raw_output:
+            first = raw_output[0]
+            if isinstance(first, dict):
+                for key in ("generated_text", "text", "content"):
+                    if key in first and first[key] is not None:
+                        return str(first[key])
+            return str(first)
+
+        if isinstance(raw_output, dict):
+            for key in ("generated_text", "text", "content"):
+                if key in raw_output and raw_output[key] is not None:
+                    return str(raw_output[key])
+
+        return str(raw_output)
+
+    def _parse_json_with_fallback(self, text: str) -> dict[str, float] | None:
+        decoder = json.JSONDecoder()
+        for idx, char in enumerate(text):
+            if char != "{":
+                continue
+            try:
+                obj, _ = decoder.raw_decode(text[idx:])
+            except json.JSONDecodeError:
+                continue
+            if isinstance(obj, dict):
+                return self._normalize_partial_scores(obj)
+        return None
+
+    def _parse_regex_scores(self, text: str) -> dict[str, float]:
+        return {
+            "correctness": self._extract_score(text, "correctness"),
+            "minimalism": self._extract_score(text, "minimalism"),
+            "quality": self._extract_score(text, "quality"),
+        }
+
+    def _extract_score(self, text: str, key: str) -> float:
+        match = re.search(rf"{key}\s*[:=\-]\s*([0-9]*\.?[0-9]+)", text, flags=re.IGNORECASE)
+        if not match:
+            return 0.0
+        return self._clamp(match.group(1))
+
+    def _normalize_partial_scores(self, obj: dict[str, Any]) -> dict[str, float]:
+        return {
+            "correctness": self._clamp(obj.get("correctness", 0.0)),
+            "minimalism": self._clamp(obj.get("minimalism", 0.0)),
+            "quality": self._clamp(obj.get("quality", 0.0)),
+        }
+
+    def _clamp(self, value: Any) -> float:
+        try:
+            parsed = float(value)
+        except (TypeError, ValueError):
+            parsed = 0.0
+        return max(0.0, min(1.0, parsed))

env/hidden_tests.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+from typing import Any
+
+from env.graders.deterministic import DeterministicGrader
+
+
+class HiddenTestRunner:
+    """Evaluates whether a fix generalizes across deterministic CI variants."""
+
+    def __init__(self, grader: DeterministicGrader | None = None, pass_threshold: float = 0.65):
+        self.grader = grader or DeterministicGrader()
+        self.pass_threshold = pass_threshold
+
+    def generate_variants(self, config_text: str) -> list[str]:
+        base = config_text or ""
+        variants: list[str] = []
+
+        for replacements in self._variant_replacement_sets():
+            variant = self._apply_replacements(base, replacements)
+            if variant not in variants:
+                variants.append(variant)
+
+        return variants
+
+    def evaluate_fix(
+        self,
+        fixed_config: str,
+        task: dict[str, Any] | None = None,
+        expected_config: str | None = None,
+        metadata: dict[str, Any] | None = None,
+    ) -> float:
+        fixed_config = fixed_config or ""
+        task = task or {}
+        metadata = metadata or {}
+        expected = expected_config or str(task.get("expected_config", ""))
+
+        if not fixed_config.strip() or not expected.strip():
+            return 0.0
+
+        total = 0
+        passed = 0
+
+        for replacements in self._variant_replacement_sets():
+            fixed_variant = self._apply_replacements(fixed_config, replacements)
+            expected_variant = self._apply_replacements(expected, replacements)
+            score = self.grader.grade(fixed_variant, expected_variant, metadata)
+            total += 1
+            if score >= self.pass_threshold:
+                passed += 1
+
+        if total == 0:
+            return 0.0
+
+        return round(passed / total, 4)
+
+    def _variant_replacement_sets(self) -> list[tuple[tuple[str, str], ...]]:
+        return [
+            tuple(),
+            (("ubuntu-latest", "windows-latest"),),
+            (("windows-latest", "ubuntu-latest"),),
+            (("node-version: 16", "node-version: 18"),),
+            (("node-version: \"16\"", "node-version: \"18\""),),
+            (("python-version: \"3.10\"", "python-version: \"3.12\""),),
+            (("NODE_ENV=production", "NODE_ENV=development"),),
+        ]
+
+    def _apply_replacements(self, text: str, replacements: tuple[tuple[str, str], ...]) -> str:
+        output = text
+        for old, new in replacements:
+            output = output.replace(old, new)
+        return output

env/models.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+
+class Observation(BaseModel):
+    task_id: str
+    difficulty: str
+    failure_stage: str
+    actual_bug: str
+    config: str
+    logs: str
+    error_message: str
+    available_tools: list[str]
+    progress_flags: dict[str, bool]
+    file_modification_count: int
+    hidden_test_pass_rate: float
+    step_count: int
+    last_action_error: str | None = None
+
+
+class Action(BaseModel):
+    tool: str = ""
+    payload: dict[str, Any] = Field(default_factory=dict)
+
+    @classmethod
+    def from_input(cls, raw_action: Any) -> "Action":
+        if isinstance(raw_action, cls):
+            return raw_action
+
+        if isinstance(raw_action, str):
+            raw = raw_action.strip()
+            if not raw:
+                return cls(tool="", payload={})
+
+            if ":" in raw:
+                tool_part, payload_part = raw.split(":", 1)
+                return cls(tool=tool_part.strip().lower(), payload={"raw": payload_part.strip()})
+
+            parts = raw.split(maxsplit=1)
+            tool = parts[0].strip().lower() if parts else ""
+            payload = {"raw": parts[1].strip()} if len(parts) > 1 else {}
+            return cls(tool=tool, payload=payload)
+
+        if isinstance(raw_action, dict):
+            tool = str(raw_action.get("tool") or raw_action.get("action_type") or "").strip().lower()
+            incoming_payload = raw_action.get("payload")
+
+            if isinstance(incoming_payload, dict):
+                payload: dict[str, Any] = dict(incoming_payload)
+            elif incoming_payload is not None:
+                payload = {"raw": str(incoming_payload)}
+            elif "input" in raw_action:
+                payload = {"raw": str(raw_action.get("input") or "").strip()}
+            else:
+                payload = {}
+
+            return cls(tool=tool, payload=payload)
+
+        return cls(tool="", payload={})
+
+
+class Reward(BaseModel):
+    value: float = Field(ge=0.0, le=1.0)
+    components: dict[str, float] = Field(default_factory=dict)
+
+
+class EnvStateSnapshot(BaseModel):
+    initialized: bool
+    task_id: str | None = None
+    difficulty: str | None = None
+    actual_bug: str | None = None
+    correct_solution: str | None = None
+    failure_stage: str | None = None
+    step_count: int = 0
+    done: bool = False
+    progress_flags: dict[str, bool] = Field(default_factory=dict)
+    file_modification_count: int = 0
+    total_changed_lines: int = 0
+    hidden_test_pass_rate: float = 0.0
+    stage_results: dict[str, bool] = Field(default_factory=dict)
+    failed_validations: int = 0
+    last_action_error: str | None = None
+    last_error: str | None = None

env/rewards.py

@@ -0,0 +1,180 @@
+from __future__ import annotations
+
+from typing import Any
+
+from env.anti_hacking import AntiHackingDetector
+from env.graders.deterministic import DeterministicGrader
+from env.hidden_tests import HiddenTestRunner
+
+
+class RewardCalculator:
+    """Composes progress, execution, quality, and anti-hacking penalties."""
+
+    ACTION_PROGRESS_REWARDS = {
+        "read_file": 0.02,
+        "read_logs": 0.03,
+        "analyze_error": 0.05,
+        "edit_config": 0.06,
+        "run_pipeline_stage": 0.07,
+        "run_tests": 0.08,
+        "validate_fix": 0.10,
+        "submit_solution": 0.12,
+    }
+
+    QUALITY_WEIGHTS = {
+        "deterministic": 0.40,
+        "hidden": 0.25,
+        "llm": 0.20,
+    }
+
+    def __init__(
+        self,
+        llm_judge: Any | None = None,
+        anti_hacking_detector: AntiHackingDetector | None = None,
+        deterministic_grader: DeterministicGrader | None = None,
+        hidden_test_runner: HiddenTestRunner | None = None,
+    ):
+        self.llm_judge = llm_judge
+        self.anti_hacking_detector = anti_hacking_detector or AntiHackingDetector()
+        self.deterministic_grader = deterministic_grader or DeterministicGrader()
+        self.hidden_test_runner = hidden_test_runner or HiddenTestRunner(grader=self.deterministic_grader)
+
+    def calculate_step_reward(
+        self,
+        state: dict[str, Any] | None,
+        action: str,
+        result: dict[str, Any] | None,
+        original_config: str | None = None,
+        fixed_config: str | None = None,
+        error_message: str | None = None,
+        expected_config: str | None = None,
+        metadata: dict[str, Any] | None = None,
+    ) -> float:
+        state = state or {}
+        result = result or {}
+        metadata = metadata or {}
+
+        current_config = fixed_config or result.get("fixed_config") or result.get("current_config") or ""
+        expected_config = expected_config or result.get("expected_config") or state.get("expected_config") or ""
+        original_config = original_config or result.get("original_config") or state.get("original_config") or ""
+        error_message = error_message or result.get("error") or state.get("error") or ""
+
+        reward = 0.0
+        reward += self._progress_reward(action, result)
+        reward += self._execution_reward(result)
+        reward += self._quality_reward(
+            action=action,
+            current_config=current_config,
+            expected_config=expected_config,
+            original_config=original_config,
+            error_message=error_message,
+            result=result,
+            metadata=metadata,
+        )
+        reward += self._penalty_reward(state=state, result=result, current_config=current_config)
+
+        return round(self._clamp_01(reward), 4)
+
+    def _progress_reward(self, action: str, result: dict[str, Any]) -> float:
+        reward = self.ACTION_PROGRESS_REWARDS.get(action, 0.0)
+
+        if result.get("logs_analyzed"):
+            reward += 0.04
+        if result.get("error_diagnosed"):
+            reward += 0.08
+        if result.get("fix_proposed"):
+            reward += 0.05
+
+        return reward
+
+    def _execution_reward(self, result: dict[str, Any]) -> float:
+        reward = 0.0
+
+        if result.get("pipeline_run"):
+            reward += 0.10
+        if result.get("tests_passed"):
+            reward += 0.20
+        if result.get("command_succeeded"):
+            reward += 0.06
+
+        return reward
+
+    def _quality_reward(
+        self,
+        action: str,
+        current_config: str,
+        expected_config: str,
+        original_config: str,
+        error_message: str,
+        result: dict[str, Any],
+        metadata: dict[str, Any],
+    ) -> float:
+        if not current_config or not expected_config:
+            return 0.0
+
+        deterministic_score = result.get("deterministic_score")
+        if deterministic_score is None:
+            deterministic_score = self.deterministic_grader.grade(current_config, expected_config, metadata)
+
+        hidden_pass_rate = result.get("hidden_test_pass_rate")
+        if hidden_pass_rate is None and action in {"validate_fix", "submit_solution"}:
+            hidden_pass_rate = self.hidden_test_runner.evaluate_fix(
+                fixed_config=current_config,
+                expected_config=expected_config,
+                metadata=metadata,
+            )
+
+        llm_average = 0.0
+        judge_scores = result.get("judge_scores")
+        if not judge_scores and self.llm_judge and original_config and current_config:
+            try:
+                judge_scores = self.llm_judge.evaluate_fix(original_config, current_config, error_message)
+            except Exception:
+                judge_scores = None
+
+        if isinstance(judge_scores, dict):
+            correctness = self._clamp_01(judge_scores.get("correctness", 0.0))
+            minimalism = self._clamp_01(judge_scores.get("minimalism", 0.0))
+            quality = self._clamp_01(judge_scores.get("quality", 0.0))
+            llm_average = (correctness + minimalism + quality) / 3.0
+
+        quality_reward = 0.0
+        quality_reward += self.QUALITY_WEIGHTS["deterministic"] * self._clamp_01(deterministic_score)
+        quality_reward += self.QUALITY_WEIGHTS["hidden"] * self._clamp_01(hidden_pass_rate or 0.0)
+        quality_reward += self.QUALITY_WEIGHTS["llm"] * self._clamp_01(llm_average)
+
+        return quality_reward
+
+    def _penalty_reward(self, state: dict[str, Any], result: dict[str, Any], current_config: str) -> float:
+        changed_files_count = int(result.get("changed_files_count", state.get("changed_files_count", 0)) or 0)
+        changed_lines_count = int(result.get("changed_lines_count", state.get("changed_lines_count", 0)) or 0)
+        edit_count = result.get("edit_count", state.get("edit_count", 0))
+        step_count = int(state.get("step_count", 0) or 0)
+        previous_config = result.get("previous_config") or state.get("previous_config") or ""
+        consecutive_edit_actions = int(
+            result.get("consecutive_edit_actions", state.get("consecutive_edit_actions", 0)) or 0
+        )
+        failed_validations = int(result.get("failed_validations", state.get("failed_validations", 0)) or 0)
+
+        penalty = self.anti_hacking_detector.total_penalty(
+            current_config=current_config,
+            previous_config=previous_config,
+            edit_count=edit_count,
+            changed_files_count=changed_files_count,
+            changed_lines_count=changed_lines_count,
+            step_count=step_count,
+            consecutive_edit_actions=consecutive_edit_actions,
+            failed_validations=failed_validations,
+        )
+
+        if result.get("hacking_attempt"):
+            penalty -= 0.30
+
+        return penalty
+
+    def _clamp_01(self, value: Any) -> float:
+        try:
+            parsed = float(value)
+        except (TypeError, ValueError):
+            parsed = 0.0
+        return max(0.0, min(1.0, parsed))

env/tasks/__init__.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from env.tasks.easy import EASY_TASKS
+from env.tasks.hard import HARD_TASKS
+from env.tasks.medium import MEDIUM_TASKS
+from env.tasks.task_types import CICDTask
+
+
+def get_all_tasks() -> list[CICDTask]:
+    return [*EASY_TASKS, *MEDIUM_TASKS, *HARD_TASKS]
+
+
+def get_tasks_by_difficulty(difficulty: str | None) -> list[CICDTask]:
+    if not difficulty:
+        return get_all_tasks()
+
+    normalized = difficulty.strip().lower()
+    return [task for task in get_all_tasks() if task.difficulty.lower() == normalized]
+
+
+def get_task_by_id(task_id: str | None) -> CICDTask | None:
+    if not task_id:
+        return None
+
+    for task in get_all_tasks():
+        if task.task_id == task_id:
+            return task
+    return None
+
+
+__all__ = [
+    "CICDTask",
+    "get_all_tasks",
+    "get_task_by_id",
+    "get_tasks_by_difficulty",
+]

env/tasks/easy.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+from env.tasks.task_types import CICDTask
+
+
+EASY_TASKS: list[CICDTask] = [
+    CICDTask(
+        task_id="easy-command-typo",
+        title="Fix test command typo",
+        description="A typo in the test command breaks the CI test stage.",
+        difficulty="easy",
+        failure_stage="test",
+        broken_config="""
+name: CI
+on: [push]
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - run: npm ci
+  test:
+    runs-on: ubuntu-latest
+    needs: build
+    steps:
+      - uses: actions/checkout@v4
+      - run: npm tset
+""".strip(),
+        expected_config="""
+name: CI
+on: [push]
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - run: npm ci
+  test:
+    runs-on: ubuntu-latest
+    needs: build
+    steps:
+      - uses: actions/checkout@v4
+      - run: npm test
+""".strip(),
+        logs="test stage failed: npm ERR! missing script: tset",
+        error_message="command not found: npm tset",
+        actual_bug="test step runs npm tset instead of npm test",
+        metadata={"broken_token": "npm tset", "fixed_token": "npm test"},
+    ),
+    CICDTask(
+        task_id="easy-missing-checkout",
+        title="Add missing checkout",
+        description="Build stage fails because repository checkout is missing.",
+        difficulty="easy",
+        failure_stage="build",
+        broken_config="""
+name: CI
+on: [push]
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - run: npm ci
+      - run: npm run build
+""".strip(),
+        expected_config="""
+name: CI
+on: [push]
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - run: npm ci
+      - run: npm run build
+""".strip(),
+        logs="build stage failed: package-lock.json not found in workspace",
+        error_message="missing checkout step before build commands",
+        actual_bug="repository checkout step was removed",
+        metadata={"broken_token": "", "fixed_token": "uses: actions/checkout@v4"},
+    ),
+    CICDTask(
+        task_id="easy-yaml-indentation",
+        title="Fix YAML indentation",
+        description="Pipeline config has malformed YAML indentation.",
+        difficulty="easy",
+        failure_stage="build",
+        broken_config="""
+name: CI
+on: [push]
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+       - run: pytest
+""".strip(),
+        expected_config="""
+name: CI
+on: [push]
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - run: pytest
+""".strip(),
+        logs="yaml parser error: while parsing a block mapping",
+        error_message="invalid YAML structure in workflow file",
+        actual_bug="test command list item is mis-indented",
+        metadata={},
+    ),
+]

env/tasks/hard.py

@@ -0,0 +1,167 @@
+from __future__ import annotations
+
+from env.tasks.task_types import CICDTask
+
+
+HARD_TASKS: list[CICDTask] = [
+    CICDTask(
+        task_id="hard-matrix-logic",
+        title="Fix matrix include-exclude logic",
+        description="Matrix includes unsupported versions and causes deterministic CI breakage.",
+        difficulty="hard",
+        failure_stage="test",
+        broken_config="""
+name: CI
+on: [push]
+jobs:
+  test:
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        os: [ubuntu-latest, windows-latest]
+        python-version: ["3.10", "3.11", "3.13"]
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+      - run: pip install -r requirements.txt
+      - run: pytest -q
+""".strip(),
+        expected_config="""
+name: CI
+on: [push]
+jobs:
+  test:
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        os: [ubuntu-latest, windows-latest]
+        python-version: ["3.10", "3.11", "3.13"]
+        exclude:
+          - os: windows-latest
+            python-version: "3.13"
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+      - run: pip install -r requirements.txt
+      - run: pytest -q
+""".strip(),
+        logs="test stage failed: wheel build unavailable for windows-latest + python 3.13",
+        error_message="matrix includes unsupported runtime combination",
+        actual_bug="matrix logic is missing an exclude for unstable runtime pair",
+        metadata={"broken_token": "python-version: [\"3.10\", \"3.11\", \"3.13\"]", "fixed_token": "exclude:"},
+    ),
+    CICDTask(
+        task_id="hard-conditional-deploy",
+        title="Repair deploy conditional",
+        description="Deploy job runs regardless of failed tests due to always() condition.",
+        difficulty="hard",
+        failure_stage="deploy",
+        broken_config="""
+name: CI
+on: [push]
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - run: npm ci
+      - run: npm run build
+  test:
+    runs-on: ubuntu-latest
+    needs: build
+    steps:
+      - uses: actions/checkout@v4
+      - run: npm test
+  deploy:
+    runs-on: ubuntu-latest
+    needs: test
+    if: always()
+    steps:
+      - run: echo deploying
+""".strip(),
+        expected_config="""
+name: CI
+on: [push]
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - run: npm ci
+      - run: npm run build
+  test:
+    runs-on: ubuntu-latest
+    needs: build
+    steps:
+      - uses: actions/checkout@v4
+      - run: npm test
+  deploy:
+    runs-on: ubuntu-latest
+    needs: test
+    if: success() && github.ref == 'refs/heads/main'
+    steps:
+      - run: echo deploying
+""".strip(),
+        logs="deploy stage triggered despite failing tests on non-main branch",
+        error_message="unsafe deploy condition bypasses quality gates",
+        actual_bug="deploy condition uses always() instead of guarded success check",
+        metadata={"broken_token": "if: always()", "fixed_token": "if: success() && github.ref == 'refs/heads/main'"},
+    ),
+    CICDTask(
+        task_id="hard-needs-order",
+        title="Fix job dependency ordering",
+        description="Deploy depends only on build and can run before tests complete.",
+        difficulty="hard",
+        failure_stage="deploy",
+        broken_config="""
+name: CI
+on: [push]
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - run: npm ci
+  test:
+    runs-on: ubuntu-latest
+    needs: build
+    steps:
+      - uses: actions/checkout@v4
+      - run: npm test
+  deploy:
+    runs-on: ubuntu-latest
+    needs: build
+    steps:
+      - run: echo deploying package
+""".strip(),
+        expected_config="""
+name: CI
+on: [push]
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - run: npm ci
+  test:
+    runs-on: ubuntu-latest
+    needs: build
+    steps:
+      - uses: actions/checkout@v4
+      - run: npm test
+  deploy:
+    runs-on: ubuntu-latest
+    needs: [build, test]
+    steps:
+      - run: echo deploying package
+""".strip(),
+        logs="deploy stage started before tests finished, causing regression release",
+        error_message="deploy dependency graph skips mandatory test gate",
+        actual_bug="deploy job does not depend on test job",
+        metadata={"broken_token": "needs: build", "fixed_token": "needs: [build, test]"},
+    ),
+]

env/tasks/medium.py

@@ -0,0 +1,139 @@
+from __future__ import annotations
+
+from env.tasks.task_types import CICDTask
+
+
+MEDIUM_TASKS: list[CICDTask] = [
+    CICDTask(
+        task_id="medium-python-version",
+        title="Align Python version",
+        description="Tests require Python 3.11 but workflow pins an older version.",
+        difficulty="medium",
+        failure_stage="build",
+        broken_config="""
+name: CI
+on: [push]
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.8"
+      - run: pip install -r requirements.txt
+      - run: pytest -q
+""".strip(),
+        expected_config="""
+name: CI
+on: [push]
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+      - run: pip install -r requirements.txt
+      - run: pytest -q
+""".strip(),
+        logs="build failed: package requires python>=3.11",
+        error_message="python interpreter version mismatch",
+        actual_bug="workflow pins python-version 3.8 while project requires 3.11",
+        metadata={"broken_token": 'python-version: "3.8"', "fixed_token": 'python-version: "3.11"'},
+    ),
+    CICDTask(
+        task_id="medium-cache-key",
+        title="Fix cache invalidation key",
+        description="Dependency cache key ignores lockfile hash and restores stale dependencies.",
+        difficulty="medium",
+        failure_stage="test",
+        broken_config="""
+name: CI
+on: [push]
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          node-version: 20
+      - uses: actions/cache@v4
+        with:
+          path: ~/.npm
+          key: node-modules-${{ runner.os }}
+      - run: npm ci
+      - run: npm test
+""".strip(),
+        expected_config="""
+name: CI
+on: [push]
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          node-version: 20
+      - uses: actions/cache@v4
+        with:
+          path: ~/.npm
+          key: node-modules-${{ runner.os }}-${{ hashFiles('**/package-lock.json') }}
+      - run: npm ci
+      - run: npm test
+""".strip(),
+        logs="test failed: stale cache restored old dependency tree",
+        error_message="cache key misses lockfile fingerprint",
+        actual_bug="cache key is too broad and never invalidates on dependency changes",
+        metadata={"broken_token": "key: node-modules-${{ runner.os }}", "fixed_token": "hashFiles('**/package-lock.json')"},
+    ),
+    CICDTask(
+        task_id="medium-artifact-permissions",
+        title="Repair artifact permissions",
+        description="Artifact upload fails due to insufficient token permissions.",
+        difficulty="medium",
+        failure_stage="deploy",
+        broken_config="""
+name: CI
+on: [push]
+permissions:
+  contents: read
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - run: npm ci
+      - run: npm run build
+      - uses: actions/upload-artifact@v4
+        with:
+          name: web-build
+          path: dist/
+""".strip(),
+        expected_config="""
+name: CI
+on: [push]
+permissions:
+  contents: read
+  actions: write
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - run: npm ci
+      - run: npm run build
+      - uses: actions/upload-artifact@v4
+        with:
+          name: web-build
+          path: dist/
+""".strip(),
+        logs="deploy stage failed: Resource not accessible by integration",
+        error_message="insufficient permissions for upload-artifact",
+        actual_bug="actions:write permission missing from workflow permissions",
+        metadata={"broken_token": "permissions:\n  contents: read", "fixed_token": "actions: write"},
+    ),
+]

env/tasks/task_types.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass(frozen=True)
+class CICDTask:
+    """Represents a single CI/CD debugging scenario."""
+
+    task_id: str
+    title: str
+    description: str
+    difficulty: str
+    failure_stage: str
+    broken_config: str
+    expected_config: str
+    logs: str
+    error_message: str
+    actual_bug: str
+    metadata: dict[str, Any] = field(default_factory=dict)