env/anti_hacking.py

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)