| 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) |
|
|
