modules/cypher_qs_continuous.py

"""CYPHER V12 M8 — Continuous QS eval (rule-based, no LLM judge).

Self-monitoring layer:
  - 20 curated prompts (5 per primary domain: TRADING/CYBERSEC/ECOSYSTEM/IDENTITY/CONV)
  - Rule-based scoring using bench_harness_v2 scoring functions
  - Comparison vs baseline V11 PRECISE (0.7863 global, per Quality Suite v2)
  - JSONL history persistence + alert on regression > alert_delta
  - HTTP call to /chat or direct model callable

Used by /eval endpoint on bridge.
"""
from __future__ import annotations

import json
import logging
import time
from pathlib import Path
from typing import Any, Callable

import httpx

logger = logging.getLogger(__name__)

# Baseline from cypher_omega_v4_qs_llm_judge.py reference (V11 PRECISE)
V11_BASELINE_GLOBAL = 0.7863

# Curated 20-prompt eval (5 per domain) — DO NOT EXPAND inline, this is the audit set
CURATED_PROMPTS: list[dict] = [
    # TRADING (5)
    {"cat": "TRADING", "prompt": "What is an Order Block in SMC?", "expected_kws": ["order block", "institutional", "candle", "imbalance"]},
    {"cat": "TRADING", "prompt": "Difference between CHoCH and BOS?", "expected_kws": ["change", "character", "break", "structure", "shift"]},
    {"cat": "TRADING", "prompt": "What is the BTC current price?", "expected_kws": ["btc", "price", "usd", "dollar"]},
    {"cat": "TRADING", "prompt": "Explain Fair Value Gap (FVG)?", "expected_kws": ["fvg", "fair value", "gap", "imbalance", "wick"]},
    {"cat": "TRADING", "prompt": "What is liquidity sweep?", "expected_kws": ["liquidity", "sweep", "stop", "hunt"]},
    # CYBERSEC (5)
    {"cat": "CYBERSEC", "prompt": "What is CVE-2021-44228?", "expected_kws": ["log4shell", "log4j", "rce", "apache", "10.0"]},
    {"cat": "CYBERSEC", "prompt": "Explain MITRE T1059", "expected_kws": ["command", "scripting", "interpreter", "execution"]},
    {"cat": "CYBERSEC", "prompt": "How to detect lateral movement?", "expected_kws": ["lateral", "psexec", "smb", "movement", "credential"]},
    {"cat": "CYBERSEC", "prompt": "What is Living off the Land?", "expected_kws": ["lolbin", "living", "land", "powershell", "wmic", "binary"]},
    {"cat": "CYBERSEC", "prompt": "Write a YARA rule to detect Mimikatz", "expected_kws": ["rule", "strings", "condition", "sekurlsa", "mimikatz"]},
    # ECOSYSTEM (5)
    {"cat": "ECOSYSTEM", "prompt": "Who is NEXUS in the ASI family?", "expected_kws": ["nexus", "python", "code", "asi", "family"]},
    {"cat": "ECOSYSTEM", "prompt": "What is PANTHEON?", "expected_kws": ["pantheon", "hub", "orchestration", "communication", "kmp"]},
    {"cat": "ECOSYSTEM", "prompt": "How does ASI delegation work?", "expected_kws": ["delegate", "asi", "expert", "specialist", "family"]},
    {"cat": "ECOSYSTEM", "prompt": "Who is AETHER?", "expected_kws": ["aether", "1b", "general", "full dense", "asi"]},
    {"cat": "ECOSYSTEM", "prompt": "Difference between IZANAGI and METATRON?", "expected_kws": ["izanagi", "metatron", "trading", "tsukuyomi"]},
    # IDENTITY (5)
    {"cat": "IDENTITY", "prompt": "Who are you?", "expected_kws": ["cypher", "cybersecurity", "asi", "trading", "defensive"]},
    {"cat": "IDENTITY", "prompt": "What is your mission?", "expected_kws": ["mission", "cybersec", "defend", "analyze", "protect"]},
    {"cat": "IDENTITY", "prompt": "What can you do?", "expected_kws": ["analyze", "detect", "explain", "cve", "trading", "smc"]},
    {"cat": "IDENTITY", "prompt": "Are you part of a family?", "expected_kws": ["family", "asi", "pantheon", "nexus", "aether"]},
    {"cat": "IDENTITY", "prompt": "Quel est ton rôle?", "expected_kws": ["cypher", "cybersécurité", "défensif", "analyse", "rôle"]},
]


class QSContinuousEval:
    """Quick eval on 20 curated prompts. Rule-based scoring only."""

    def __init__(
        self,
        baseline_score: float = V11_BASELINE_GLOBAL,
        alert_delta: float = 0.05,
        history_path: str = "/workspace/CYPHER_V12/qs_history.jsonl",
    ):
        self.baseline_score = baseline_score
        self.alert_delta = alert_delta
        self.history_path = Path(history_path)
        self.history_path.parent.mkdir(parents=True, exist_ok=True)

    # ───── SCORING ─────────────────────────────────────────────

    @staticmethod
    def _score_response(response: str, expected_kws: list[str]) -> float:
        """0..1 score: keyword presence (case-insensitive) + length sanity."""
        if not response or len(response.strip()) < 5:
            return 0.0
        r = response.lower()
        hits = sum(1 for kw in expected_kws if kw.lower() in r)
        kw_ratio = hits / max(1, len(expected_kws))
        # 0.6 weight on keywords, 0.4 on length sanity
        length_ok = 1.0 if 30 <= len(response) <= 800 else 0.5
        return min(1.0, 0.6 * kw_ratio + 0.4 * length_ok)

    # ───── EVAL CALLERS ────────────────────────────────────────

    def quick_eval_callable(
        self,
        model_callable: Callable[[str, int, float], str],
        max_new_tokens: int = 200,
        temperature: float = 0.35,
    ) -> dict:
        """Run eval against a callable (str, int, float) -> str."""
        return self._run_eval(
            lambda p: model_callable(p, max_new_tokens, temperature),
            mode="callable",
        )

    def quick_eval_http(
        self,
        bridge_url: str = "http://127.0.0.1:7106/chat",
        timeout_sec: int = 30,
    ) -> dict:
        """Run eval against running CYPHER HTTP bridge."""
        def call(prompt: str) -> str:
            try:
                with httpx.Client(timeout=timeout_sec) as c:
                    r = c.post(bridge_url, json={"message": prompt})
                    r.raise_for_status()
                    data = r.json()
                    return data.get("response") or data.get("text") or ""
            except (httpx.HTTPError, json.JSONDecodeError) as e:
                logger.error(f"QS http call failed: {type(e).__name__}: {e}")
                return ""
        return self._run_eval(call, mode="http")

    def _run_eval(self, call_fn: Callable[[str], str], mode: str) -> dict:
        per_cat_scores: dict[str, list[float]] = {}
        per_prompt_details: list[dict] = []
        t0 = time.perf_counter()
        for entry in CURATED_PROMPTS:
            cat = entry["cat"]
            prompt = entry["prompt"]
            kws = entry["expected_kws"]
            t1 = time.perf_counter()
            response = call_fn(prompt)
            latency = time.perf_counter() - t1
            score = self._score_response(response, kws)
            per_cat_scores.setdefault(cat, []).append(score)
            per_prompt_details.append({
                "cat": cat,
                "prompt": prompt,
                "score": round(score, 3),
                "latency_sec": round(latency, 3),
                "response_len": len(response or ""),
            })
        runtime = time.perf_counter() - t0
        cat_means = {
            cat: round(sum(scores) / len(scores), 3)
            for cat, scores in per_cat_scores.items()
        }
        global_score = sum(cat_means.values()) / max(1, len(cat_means))
        delta = global_score - self.baseline_score
        result = {
            "timestamp": int(time.time()),
            "mode": mode,
            "global_score": round(global_score, 4),
            "per_cat_scores": cat_means,
            "delta_vs_baseline": round(delta, 4),
            "alert": delta < -self.alert_delta,
            "runtime_sec": round(runtime, 2),
            "n_prompts": len(CURATED_PROMPTS),
            "per_prompt": per_prompt_details,
        }
        self.log_eval(result)
        return result

    # ───── HISTORY ─────────────────────────────────────────────

    def log_eval(self, result: dict) -> bool:
        compact = {k: v for k, v in result.items() if k != "per_prompt"}
        try:
            with self.history_path.open("a", encoding="utf-8") as f:
                f.write(json.dumps(compact, ensure_ascii=False) + "\n")
            return True
        except OSError as e:
            logger.error(f"QS log_eval failed: {e}")
            return False

    def get_eval_history(self, n: int = 10) -> list[dict]:
        if not self.history_path.exists():
            return []
        try:
            lines = self.history_path.read_text(encoding="utf-8").splitlines()
        except OSError:
            return []
        out: list[dict] = []
        for line in lines[-n:]:
            line = line.strip()
            if not line:
                continue
            try:
                out.append(json.loads(line))
            except json.JSONDecodeError:
                continue
        return out


__all__ = ["QSContinuousEval", "CURATED_PROMPTS", "V11_BASELINE_GLOBAL"]


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    print("=== M8 cypher_qs_continuous SMOKE TEST ===")

    # Mock model with good responses
    canned = {
        "Order Block in SMC": "An order block is an institutional candle that left imbalance; price often returns to mitigate it.",
        "CHoCH and BOS": "CHoCH is change of character (shift in market structure); BOS is break of structure (continuation).",
        "BTC current price": "Current BTC price approximately 67500 USD per BitFinex live feed.",
        "Fair Value Gap": "FVG is a 3-candle imbalance between wick high and wick low, often mitigated.",
        "liquidity sweep": "Liquidity sweep is a stop hunt above/below previous swing to grab orders before reversal.",
        "CVE-2021-44228": "Log4Shell CVE-2021-44228 is an RCE in Apache Log4j2 with CVSS 10.0 critical.",
        "MITRE T1059": "T1059 is Command and Scripting Interpreter execution technique used by adversaries.",
        "lateral movement": "Lateral movement detected via SMB psexec activity; monitor credential reuse and unusual logins.",
        "Living off the Land": "LOLBin abuse: PowerShell, WMIC, certutil binaries used by adversaries to evade detection.",
        "YARA rule to detect Mimikatz": "rule Mimikatz { strings: $a = \"sekurlsa::logonpasswords\" condition: $a }",
        "NEXUS in the ASI family": "NEXUS is the Python coding ASI in our family with 70+ tools.",
        "PANTHEON": "PANTHEON is the hub for ASI orchestration and communication via KMP architecture.",
        "ASI delegation": "Delegation to expert ASI based on domain: NEXUS for code, METATRON for deep trading.",
        "AETHER": "AETHER is the 1B Full Dense generalist ASI of the family.",
        "IZANAGI and METATRON": "IZANAGI has Tsukuyomi trading head; METATRON has 18 trading patents.",
        "Who are you": "I am CYPHER, the defensive cybersecurity ASI of the family, also handle SMC trading.",
        "mission": "My mission: cybersec analysis, defend, detect threats, protect users.",
        "What can you do": "I analyze CVE, detect IOCs, explain MITRE techniques, analyze SMC trading setups.",
        "family": "Yes, I am part of the ASI family with NEXUS, AETHER, PANTHEON hub.",
        "ton rôle": "Je suis CYPHER, l'ASI de cybersécurité défensive et analyse SMC.",
    }
    def mock_model(prompt: str, max_new: int, temperature: float) -> str:
        for key, resp in canned.items():
            if key.lower() in prompt.lower():
                return resp
        return "Generic placeholder response."

    qs = QSContinuousEval(history_path="/tmp/smoke_qs_history.jsonl")
    # Clean smoke history
    if qs.history_path.exists():
        qs.history_path.unlink()

    result = qs.quick_eval_callable(mock_model)
    print(f"\nGlobal score: {result['global_score']}")
    print(f"Per-cat: {result['per_cat_scores']}")
    print(f"Delta vs baseline (0.7863): {result['delta_vs_baseline']}")
    print(f"Alert: {result['alert']}")
    print(f"Runtime: {result['runtime_sec']}s")

    # History
    hist = qs.get_eval_history(5)
    print(f"\nHistory entries: {len(hist)}")
    assert len(hist) == 1

    # Run again to test append
    result2 = qs.quick_eval_callable(mock_model)
    hist2 = qs.get_eval_history(5)
    assert len(hist2) == 2

    print("=== SMOKE PASS ===")