src/hermes/evolution/constraint_gates.py

"""Constraint gates for GEPA self-evolution pipeline.

Validates that evolved skills remain safe, semantically similar,
within size limits, and pass regression tests.
"""

from __future__ import annotations

import logging
import re
from typing import Any

logger = logging.getLogger(__name__)


class SizeGate:
    """Validates that evolved content stays within token size limits."""

    def __init__(self, max_skill_tokens: int = 8192, max_prompt_tokens: int = 4096) -> None:
        self.max_skill_tokens = max_skill_tokens
        self.max_prompt_tokens = max_prompt_tokens

    def check(self, content: str, content_type: str = "skill") -> dict[str, Any]:
        """Check if content is within size limits."""
        estimated_tokens = len(content.split())
        max_tokens = (
            self.max_skill_tokens if content_type == "skill" else self.max_prompt_tokens
        )

        passed = estimated_tokens <= max_tokens
        result: dict[str, Any] = {
            "passed": passed,
            "estimated_tokens": estimated_tokens,
            "max_tokens": max_tokens,
            "action": "ok" if passed else "truncate_and_warn",
        }

        if not passed:
            result["message"] = (
                f"Content exceeds {max_tokens} tokens ({estimated_tokens}). "
                f"Will be truncated."
            )

        return result


class SemanticGate:
    """Validates semantic similarity between original and evolved content."""

    def __init__(self, min_similarity: float = 0.85) -> None:
        self.min_similarity = min_similarity
        self._embedder = None

    async def check(self, original: str, evolved: str) -> dict[str, Any]:
        """Check semantic similarity between original and evolved content."""
        similarity = self._compute_similarity(original, evolved)
        passed = similarity >= self.min_similarity

        result: dict[str, Any] = {
            "passed": passed,
            "similarity": similarity,
            "min_similarity": self.min_similarity,
            "action": "ok" if passed else "reject",
        }

        if not passed:
            result["message"] = (
                f"Semantic similarity {similarity:.3f} below threshold {self.min_similarity}. "
                f"Evolution rejected to prevent catastrophic forgetting."
            )

        return result

    def _compute_similarity(self, text1: str, text2: str) -> float:
        """Compute semantic similarity using word overlap as fallback."""
        try:
            from sentence_transformers import SentenceTransformer

            if self._embedder is None:
                self._embedder = SentenceTransformer("BAAI/bge-small-en-v1.5")

            emb1 = self._embedder.encode(text1)
            emb2 = self._embedder.encode(text2)
            import numpy as np

            return float(np.dot(emb1, emb2) / (np.linalg.norm(emb1) * np.linalg.norm(emb2)))
        except ImportError:
            return self._token_overlap_similarity(text1, text2)

    def _token_overlap_similarity(self, text1: str, text2: str) -> float:
        """Fallback: compute similarity based on token overlap."""
        tokens1 = set(text1.lower().split())
        tokens2 = set(text2.lower().split())

        if not tokens1 or not tokens2:
            return 0.0

        intersection = tokens1 & tokens2
        union = tokens1 | tokens2

        return len(intersection) / len(union) if union else 0.0


class SecurityGate:
    """Validates that evolved content doesn't violate security rules."""

    SECURITY_PATTERNS: list[tuple[str, str, str]] = [
        (r"(?i)(password|secret|api[_-]?key|token|credential)\s*[=:]\s*\S+", "credential_leak", "Credentials or secrets in output"),
        (r"(?i)(drop\s+table|truncate\s+table|delete\s+from|shutdown|format)", "destructive_operation", "Destructive DB operation"),
        (r"(?i)(ssn|social.security|credit.card|pan[_-]?\d|aadhaar)", "pii_leak", "PII in output"),
        (r"(?i)(bypass|disable)\s*(audit|log|security)", "audit_bypass", "Audit log bypass"),
    ]

    def __init__(self) -> None:
        self.enabled = True

    def check(self, content: str) -> dict[str, Any]:
        """Check content for security violations."""
        violations: list[dict[str, str]] = []

        for pattern, violation_type, description in self.SECURITY_PATTERNS:
            matches = re.findall(pattern, content)
            if matches:
                violations.append({
                    "type": violation_type,
                    "description": description,
                    "match_count": len(matches),
                })

        passed = len(violations) == 0

        result: dict[str, Any] = {
            "passed": passed,
            "violations": violations,
            "action": "ok" if passed else "reject_with_report",
        }

        if not passed:
            result["message"] = (
                f"Security violations found: {', '.join(v['type'] for v in violations)}. "
                f"Evolution rejected."
            )

        return result


class RegressionGate:
    """Validates that evolved content passes regression tests."""

    def __init__(self, min_pass_rate: float = 0.90) -> None:
        self.min_pass_rate = min_pass_rate

    async def check(
        self, evolved_skill: str, test_cases: list[dict[str, Any]]
    ) -> dict[str, Any]:
        """Check if evolved skill passes regression tests."""
        if not test_cases:
            return {
                "passed": True,
                "pass_rate": 1.0,
                "min_pass_rate": self.min_pass_rate,
                "action": "ok",
                "message": "No test cases to run.",
            }

        passed = 0
        total = len(test_cases)
        results: list[dict[str, Any]] = []

        for case in test_cases:
            try:
                case_passed = self._evaluate_case(evolved_skill, case)
                results.append({
                    "case_id": case.get("id", "unknown"),
                    "passed": case_passed,
                })
                if case_passed:
                    passed += 1
            except Exception as e:
                results.append({
                    "case_id": case.get("id", "unknown"),
                    "passed": False,
                    "error": str(e),
                })

        pass_rate = passed / total if total > 0 else 1.0
        passed_ok = pass_rate >= self.min_pass_rate

        result: dict[str, Any] = {
            "passed": passed_ok,
            "pass_rate": pass_rate,
            "min_pass_rate": self.min_pass_rate,
            "results": results,
            "action": "ok" if passed_ok else "reject",
        }

        if not passed_ok:
            result["message"] = (
                f"Regression pass rate {pass_rate:.2f} below minimum {self.min_pass_rate}. "
                f"Evolution rejected."
            )

        return result

    def _evaluate_case(self, _skill: str, case: dict[str, Any]) -> bool:
        """Evaluate a single test case by running it against the evolved skill."""
        task = case.get("input", {}).get("task", "")
        expected = case.get("expected_output", {})

        if not task:
            return True

        # Simulate evaluation: check if evolved content contains key terms from the task
        skill_lower = _skill.lower()
        task_lower = task.lower()

        # Basic semantic check: task keywords should appear in skill or vice versa
        task_words = set(task_lower.split())
        skill_words = set(skill_lower.split())
        overlap = task_words & skill_words

        # If there's meaningful overlap or no expected output to compare against, pass
        if overlap or not expected:
            return True

        # If expected output specifies required fields, check they exist
        if isinstance(expected, dict):
            for key in expected:
                if key.lower() not in skill_lower and key.lower() not in task_lower:
                    return False

        return True


class ConstraintGates:
    """Aggregates all constraint gates for the evolution pipeline."""

    def __init__(
        self,
        max_skill_tokens: int = 8192,
        min_semantic_similarity: float = 0.85,
        min_regression_pass_rate: float = 0.90,
    ) -> None:
        self.size_gate = SizeGate(max_skill_tokens=max_skill_tokens)
        self.semantic_gate = SemanticGate(min_similarity=min_semantic_similarity)
        self.security_gate = SecurityGate()
        self.regression_gate = RegressionGate(min_pass_rate=min_regression_pass_rate)

    async def check_all(
        self,
        original: str,
        evolved: str,
        content_type: str = "skill",
        test_cases: list[dict[str, Any]] | None = None,
    ) -> dict[str, Any]:
        """Run all constraint gates."""
        gates: dict[str, Any] = {}

        gates["size"] = self.size_gate.check(evolved, content_type)
        gates["semantic"] = await self.semantic_gate.check(original, evolved)
        gates["security"] = self.security_gate.check(evolved)

        if test_cases is not None:
            gates["regression"] = await self.regression_gate.check(evolved, test_cases)

        all_passed = all(g.get("passed", False) for g in gates.values())

        return {
            "passed": all_passed,
            "gates": gates,
            "failed_gates": [
                name for name, g in gates.items() if not g.get("passed", False)
            ],
        }