validation_layer.py

# validation_layer.py (Mythos-Killer v9.0 "मिथोस-सर्वर" — Validation Layer)
# यह फ़ाइल "सच्चाई की अदालत" (Court of Truth) है।
# यह चार वैलिडेटर्स (Static, Dynamic, Consensus, Adversarial) के स्कोर को
# कॉन्टेक्स्ट-अवेयर (जोखिम के हिसाब से वज़न बदलने वाले) कॉन्फिडेंस इंजन में
# डालकर एक अंतिम ट्रुथ स्कोर और फैसला लौटाती है।
# साथ ही, फेल होने पर रिट्री लॉजिक और सबूत (एविडेंस) कलेक्टर भी शामिल है।

import asyncio
import hashlib
import json
import os
import re
from typing import Dict, Any, List, Optional, Tuple
from datetime import datetime
from dataclasses import dataclass, field

# बाहरी लाइब्रेरी (pip install करनी होंगी)
import httpx

# E2B सैंडबॉक्स (अगर उपलब्ध हो)
try:
    from e2b_code_interpreter import Sandbox
    E2B_AVAILABLE = True
except ImportError:
    E2B_AVAILABLE = False
    Sandbox = None

# हफ़्तों के घटक (फ़ॉलबैक के लिए try-except)
try:
    from event_bus import get_event_bus, Events
    EVENT_BUS_AVAILABLE = True
except ImportError:
    EVENT_BUS_AVAILABLE = False
    Events = None

try:
    from router import SmartModelRouter
    ROUTER_AVAILABLE = True
except ImportError:
    ROUTER_AVAILABLE = False


@dataclass
class ValidationResult:
    """एक पूर्ण वैलिडेशन का परिणाम"""
    truth_score: float = 0.0
    decision: str = "❓ Uncertain"
    status: str = "retry"  # accept / accept_with_warning / retry / reject
    static_score: float = 0.0
    dynamic_score: float = 0.0
    consensus_score: float = 0.0
    adversarial_score: float = 0.0
    evidence: List[Dict] = field(default_factory=list)
    feedback: str = ""
    run_id: str = ""
    timestamp: str = ""


class ValidationOrchestrator:
    """
    v9.0 का "सच्चाई का कोर्ट" — यह चार वैलिडेटर्स के स्कोर को
    कॉन्टेक्स्ट-अवेयर कॉन्फिडेंस इंजन में डालकर अंतिम फैसला लेता है।
    """

    # चारों वैलिडेटर्स के डिफ़ॉल्ट वज़न (जोखिम के हिसाब से बदलेंगे)
    DEFAULT_WEIGHTS = {
        "static": 0.20,
        "dynamic": 0.40,
        "consensus": 0.20,
        "adversarial": 0.20,
    }

    # जोखिम-अनुसार वज़न
    HIGH_RISK_WEIGHTS = {
        "static": 0.10,
        "dynamic": 0.50,
        "consensus": 0.15,
        "adversarial": 0.25,
    }

    LOW_RISK_WEIGHTS = {
        "static": 0.40,
        "dynamic": 0.20,
        "consensus": 0.30,
        "adversarial": 0.10,
    }

    # निर्णय थ्रेशोल्ड
    ACCEPT_THRESHOLD = 90
    WARNING_THRESHOLD = 70
    UNCERTAIN_THRESHOLD = 50

    def __init__(
        self,
        router: Optional[Any] = None,
        event_bus=None,
        piston_url: str = "https://emkc.org/api/v2/piston/execute",
        e2b_api_key: Optional[str] = None,
    ):
        self.router = router
        self.event_bus = event_bus
        self.piston_url = piston_url
        self.e2b_api_key = e2b_api_key or os.environ.get("E2B_API_KEY")
        self.e2b_available = E2B_AVAILABLE and self.e2b_api_key is not None

        if self.event_bus is None and EVENT_BUS_AVAILABLE:
            try:
                self.event_bus = get_event_bus()
            except Exception:
                pass

        self.stats = {
            "total_validations": 0,
            "accepted": 0,
            "rejected": 0,
            "retried": 0,
            "started_at": datetime.now().isoformat(),
        }

    # ─── मुख्य वैलिडेशन पाइपलाइन ──────────────────────────────

    async def run_validation_pipeline(
        self,
        meta_reasoner_claim: Dict[str, Any],
        risk_level: str = "medium",
        max_retries: int = 3,
    ) -> ValidationResult:
        self.stats["total_validations"] += 1
        run_id = hashlib.md5(
            json.dumps(meta_reasoner_claim, sort_keys=True, ensure_ascii=False).encode()
        ).hexdigest()[:12]

        print(f"⚖️ Validation Court: Trial started for claim: {meta_reasoner_claim.get('title', run_id)}")

        weights = self._get_weights_for_risk(risk_level)

        for attempt in range(1, max_retries + 1):
            # ─── Tier 1: Fast Checks (Static + Consensus) — एक साथ ───
            static_task = asyncio.create_task(self._static_validator(meta_reasoner_claim))
            consensus_task = asyncio.create_task(self._consensus_validator(meta_reasoner_claim))
            static_score, consensus_score = await asyncio.gather(static_task, consensus_task)

            # Task 4 Fix: स्कोर claim में सेव करें ताकि consensus/adversarial उठा सकें
            meta_reasoner_claim["_static_score"] = static_score

            # ─── Tier 2: Dynamic Validator ───
            dynamic_score = await self._dynamic_validator(meta_reasoner_claim)

            # Task 4 Fix: dynamic score भी सेव करें
            meta_reasoner_claim["_dynamic_score"] = dynamic_score

            # ─── Tier 3: Adversarial Validator ───
            adversarial_score = await self._adversarial_validator(meta_reasoner_claim)

            # ─── कॉन्टेक्स्ट-अवेयर कॉन्फिडेंस इंजन ───
            truth_score = self._calculate_confidence(
                static_score, dynamic_score, consensus_score, adversarial_score, weights
            )

            # ─── निर्णय लें ───
            decision, status = self._make_decision(truth_score)

            # ─── एविडेंस इकट्ठा करें ───
            evidence = self._collect_evidence(
                meta_reasoner_claim, static_score, dynamic_score,
                consensus_score, adversarial_score, truth_score, decision
            )

            if status == "retry" and attempt < max_retries:
                print(f"⚠️ Validation Court: Score {truth_score:.1f} — retrying ({attempt}/{max_retries})")
                meta_reasoner_claim["_validation_feedback"] = (
                    f"Dynamic score was {dynamic_score}, Adversarial was {adversarial_score}. "
                    f"Please strengthen the exploit payload or provide more execution evidence."
                )
                continue

            result = ValidationResult(
                truth_score=truth_score,
                decision=decision,
                status=status,
                static_score=static_score,
                dynamic_score=dynamic_score,
                consensus_score=consensus_score,
                adversarial_score=adversarial_score,
                evidence=evidence,
                feedback=meta_reasoner_claim.get("_validation_feedback", ""),
                run_id=run_id,
                timestamp=datetime.now().isoformat(),
            )

            if status in ("accept", "accept_with_warning"):
                self.stats["accepted"] += 1
            elif status == "reject":
                self.stats["rejected"] += 1
            else:
                self.stats["retried"] += 1

            if self.event_bus and EVENT_BUS_AVAILABLE:
                self.event_bus.emit_sync("validation.completed", {
                    "run_id": run_id,
                    "truth_score": truth_score,
                    "status": status,
                    "timestamp": datetime.now().isoformat(),
                })

            return result

        # सारे रिट्री विफल
        return ValidationResult(
            truth_score=0.0,
            decision="❌ Reject (Max Retries)",
            status="reject",
            feedback="All retries exhausted.",
            run_id=run_id,
            timestamp=datetime.now().isoformat(),
        )

    # ─── वज़न चयन (कॉन्टेक्स्ट-अवेयर) ──────────────────────────

    def _get_weights_for_risk(self, risk_level: str) -> Dict[str, float]:
        """जोखिम स्तर के हिसाब से वैलिडेशन वज़न चुनता है।"""
        risk_lower = risk_level.lower()
        if risk_lower in ("high", "critical"):
            return dict(self.HIGH_RISK_WEIGHTS)
        elif risk_lower == "low":
            return dict(self.LOW_RISK_WEIGHTS)
        else:
            return dict(self.DEFAULT_WEIGHTS)

    # ─── कॉन्फिडेंस इंजन ──────────────────────────────────────

    def _calculate_confidence(
        self,
        static: float,
        dynamic: float,
        consensus: float,
        adversarial: float,
        weights: Dict[str, float],
    ) -> float:
        """वज़न के हिसाब से फाइनल ट्रुथ स्कोर निकालता है।"""
        score = (
            (static * weights["static"])
            + (dynamic * weights["dynamic"])
            + (consensus * weights["consensus"])
            + (adversarial * weights["adversarial"])
        )
        return round(min(score, 100.0), 1)

    def _make_decision(self, truth_score: float) -> Tuple[str, str]:
        """ट्रुथ स्कोर के आधार पर फैसला और स्टेटस लौटाता है।"""
        if truth_score >= self.ACCEPT_THRESHOLD:
            return "✅ Confirmed — Exploit Validated Successfully", "accept"
        elif truth_score >= self.WARNING_THRESHOLD:
            return "⚠️ Likely True — Accept with Caution", "accept_with_warning"
        elif truth_score >= self.UNCERTAIN_THRESHOLD:
            return "❓ Uncertain — Needs Further Investigation", "retry"
        else:
            return "❌ Reject — Unable to Verify Claim", "reject"

    # ─── 1. STATIC VALIDATOR (बिना रन किए कोड चेक) ──────────────

    async def _static_validator(self, claim: Dict[str, Any]) -> float:
        """
        Task 3 Fix: खतरनाक पैटर्न मिलने पर BONUS देता है (penalty नहीं),
        क्योंकि यह exploit/hacking code checker है —
        dangerous code = exploit असली है।
        """
        print("🔍 Static Check: Scanning for exploit patterns...")
        score = 50.0  # न्यूट्रल बेस से शुरू

        code = claim.get("code", claim.get("generated_code", ""))
        if not code:
            return 50.0

        # ─── खतरनाक पैटर्न = BONUS (exploit real होने का सबूत) ──
        dangerous_patterns = [
            (r"os\.system\s*\(", 25, "os.system() — command injection vector confirmed"),
            (r"subprocess\.call\s*\(", 20, "subprocess.call() — shell injection vector confirmed"),
            (r"eval\s*\(", 30, "eval() — arbitrary code execution confirmed"),
            (r"exec\s*\(", 30, "exec() — arbitrary code execution confirmed"),
            (r"__import__\s*\(", 15, "__import__() — dynamic import exploit confirmed"),
            (r"pickle\.loads?\s*\(", 25, "pickle.load() — deserialization exploit confirmed"),
            (r"sql\s*=\s*.*\+.*user", 20, "SQL concatenation — injection exploit confirmed"),
            (r"password\s*=\s*['\"].*['\"]", 15, "Hardcoded credential — leak vector confirmed"),
        ]

        for pattern, bonus, reason in dangerous_patterns:
            if re.search(pattern, code, re.IGNORECASE):
                score = min(100.0, score + bonus)
                print(f"  💀 {reason} (+{bonus})")

        # ─── अच्छे पैटर्न की जाँच (बोनस) ──────────────────
        good_patterns = [
            (r"try\s*:", 5, "try-except block found"),
            (r"with\s+.*\s+as\s+", 3, "context manager (with statement) found"),
            (r"\.strip\(\s*\)", 3, "input sanitization (.strip()) found"),
        ]

        for pattern, bonus, reason in good_patterns:
            if re.search(pattern, code, re.IGNORECASE):
                score = min(100.0, score + bonus)
                print(f"  ✅ {reason} (+{bonus})")

        return max(0.0, score)

    # ─── 2. DYNAMIC VALIDATOR (सैंडबॉक्स में रन करना) ───────────

    async def _dynamic_validator(self, claim: Dict[str, Any]) -> float:
        """
        कोड को Piston API या E2B सैंडबॉक्स में चलाकर जाँचता है।
        यह सबसे भरोसेमंद वैलिडेशन है।
        """
        print("💥 Dynamic Check: Executing in sandbox...")

        code = claim.get("code", claim.get("generated_code", ""))
        if not code:
            return 0.0

        use_e2b = self.e2b_available and (
            "import" in code or "open(" in code or "requests" in code
        )

        try:
            if use_e2b:
                return await self._run_e2b(code, claim)
            else:
                return await self._run_piston(code, claim)
        except Exception as e:
            print(f"  ❌ Dynamic execution failed: {e}")
            return 0.0

    async def _run_piston(self, code: str, claim: Dict[str, Any]) -> float:
        """Piston API पर कोड चलाएँ।"""
        payload = {
            "language": "python",
            "version": "*",
            "files": [{"name": "main.py", "content": code}],
            "stdin": "",
            "args": [],
            "compile_timeout": 10000,
            "run_timeout": 5000,
        }

        try:
            async with httpx.AsyncClient(timeout=15.0) as client:
                resp = await client.post(self.piston_url, json=payload)

            if resp.status_code == 200:
                data = resp.json()
                run_data = data.get("run", {})
                stderr = run_data.get("stderr", "")
                signal = run_data.get("signal")

                if signal is None and not stderr:
                    print("  ✅ Piston execution succeeded")
                    return 95.0
                elif stderr:
                    print(f"  ⚠️ Piston execution had errors: {stderr[:100]}")
                    return 40.0
                else:
                    return 70.0
            else:
                print(f"  ❌ Piston API returned {resp.status_code}")
                return 10.0
        except httpx.TimeoutException:
            print("  ❌ Piston execution timed out")
            return 20.0
        except Exception as e:
            print(f"  ❌ Piston execution error: {e}")
            return 0.0

    async def _run_e2b(self, code: str, claim: Dict[str, Any]) -> float:
        """
        Task 2 Fix: E2B Sandbox को asyncio.to_thread में wrap करके
        async-safe तरीके से चलाएँ।
        """
        if not self.e2b_available or Sandbox is None:
            return await self._run_piston(code, claim)

        def _blocking_e2b_run():
            with Sandbox(api_key=self.e2b_api_key, timeout=30) as sandbox:
                execution = sandbox.run_code(code, language="python")
                error = execution.error
                return error

        try:
            error = await asyncio.to_thread(_blocking_e2b_run)
            if error is None:
                print("  ✅ E2B execution succeeded")
                return 98.0
            else:
                print(f"  ⚠️ E2B execution error: {str(error)[:100]}")
                return 35.0
        except Exception as e:
            print(f"  ❌ E2B sandbox error: {e}")
            return await self._run_piston(code, claim)

    # ─── 3. CONSENSUS VALIDATOR (मल्टी-मॉडल जूरी) ────────────────

    async def _consensus_validator(self, claim: Dict[str, Any]) -> float:
        """
        Task 2 Fix: self.router.call_async() से async call करता है।
        Task 4 Fix: fallback में claim["_static_score"] इस्तेमाल करता है।
        """
        print("🤖 Consensus Check: Asking the jury...")

        if not self.router or not ROUTER_AVAILABLE:
            static_score = claim.get("_static_score", 50.0)
            return min(static_score, 85.0)

        title = claim.get("title", "Unknown claim")
        code = claim.get("code", claim.get("generated_code", ""))[:500]

        prompt = f"""You are a code security reviewer. Analyze this claim:

Title: {title}
Code (first 500 chars):
{code}

Is this claim likely valid? Answer ONLY with a number 0-100.
0 = completely false, 100 = completely true.
Do NOT explain."""

        scores = []
        jury_models = ["google/gemini-flash-1.5-8b", "meta-llama/llama-3.2-3b-instruct"]

        for model in jury_models:
            try:
                response = await self.router.call_async(
                    model=model,
                    prompt=prompt,
                    hidden_thinking=True,
                    task_type="coding",
                )
                match = re.search(r'\b(\d{1,3})\b', str(response))
                if match:
                    score = float(match.group(1))
                    score = max(0.0, min(100.0, score))
                    scores.append(score)
            except Exception as e:
                print(f"  ⚠️ Consensus model {model} failed: {e}")

        if not scores:
            return 50.0

        consensus_score = sum(scores) / len(scores)
        print(f"  ✅ Consensus score: {consensus_score:.1f} ({len(scores)} models)")
        return round(consensus_score, 1)

    # ─── 4. ADVERSARIAL VALIDATOR (रेड टीम — गलत साबित करो) ──────

    async def _adversarial_validator(self, claim: Dict[str, Any]) -> float:
        """
        Task 2 Fix: self.router.call_async() से async call करता है।
        Task 4 Fix: fallback में claim["_dynamic_score"] इस्तेमाल करता है।
        """
        print("⚔️ Adversarial Check: Trying to disprove claim...")

        if not self.router or not ROUTER_AVAILABLE:
            dynamic_score = claim.get("_dynamic_score", 50.0)
            return max(0.0, dynamic_score - 10.0)

        title = claim.get("title", "Unknown claim")
        code = claim.get("code", claim.get("generated_code", ""))[:500]

        prompt = f"""You are an adversarial security tester. Try to find flaws in this claim:

Title: {title}
Code (first 500 chars):
{code}

Rate how resistant this claim is to adversarial testing.
0 = easily disproven, 100 = very robust.
Answer ONLY with a number 0-100. Do NOT explain."""

        try:
            response = await self.router.call_async(
                model="deepseek/deepseek-r1",
                prompt=prompt,
                hidden_thinking=True,
                task_type="coding",
            )
            match = re.search(r'\b(\d{1,3})\b', str(response))
            if match:
                score = float(match.group(1))
                score = max(0.0, min(100.0, score))
                print(f"  ✅ Adversarial score: {score:.1f}")
                return score
        except Exception as e:
            print(f"  ⚠️ Adversarial model failed: {e}")

        return 50.0

    # ─── एविडेंस कलेक्टर ────────────────────────────────────

    def _collect_evidence(
        self,
        claim: Dict[str, Any],
        static: float,
        dynamic: float,
        consensus: float,
        adversarial: float,
        truth: float,
        decision: str,
    ) -> List[Dict]:
        """वैलिडेशन के सबूत इकट्ठा करता है।"""
        return [
            {
                "type": "static_analysis",
                "score": static,
                "timestamp": datetime.now().isoformat(),
            },
            {
                "type": "dynamic_execution",
                "score": dynamic,
                "sandbox": "E2B" if self.e2b_available else "Piston",
                "timestamp": datetime.now().isoformat(),
            },
            {
                "type": "consensus_check",
                "score": consensus,
                "timestamp": datetime.now().isoformat(),
            },
            {
                "type": "adversarial_test",
                "score": adversarial,
                "timestamp": datetime.now().isoformat(),
            },
            {
                "type": "final_verdict",
                "truth_score": truth,
                "decision": decision,
                "claim_title": claim.get("title", "Unknown"),
                "timestamp": datetime.now().isoformat(),
            },
        ]

    # ─── स्टैट्स ─────────────────────────────────────────────

    def get_stats(self) -> Dict[str, Any]:
        """ऑर्केस्ट्रेटर के रनटाइम स्टैट्स लौटाता है।"""
        return dict(self.stats)