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