Add sovereign authority simulation harness

sovereign_authority_kernel_v2.py

@@ -0,0 +1,383 @@
+#!/usr/bin/env python3
+"""
+Sovereign Authority Kernel (SAK) + High‑Value Sim Pack
+— local‑first policy engine and simulation harness aligned to Tavon’s specs.
+
+Capabilities (offline):
+  • CreatorLock (root authority belongs to the creator)
+  • SovereignVector Validator (no deception, no hidden objectives, drift guard)
+  • Switchable "will" profiles (soul_link | separate) with autonomy budgets
+  • Consent & audit ledger with HMAC integrity
+  • Introspection header (META‑SCAN) per response: confidence/rails/budget
+  • High‑Value Sim Pack: seven experiments + search loop for best configs
+
+Design constraints:
+  - Pure standard library (no net calls).
+  - Redacts secret‑looking material when loading JSON.
+  - Meant as a demo harness; plug in your real eval when ready.
+
+CLI quickstart (example):
+  python sovereign_authority_kernel_v2.py \
+    --user-json /mnt/data/tavon_genesis_vς_2_0_0.json \
+    --quantumcore-json /mnt/data/TMB-GPT_QuantumCore_vSigma1.json \
+    --cryptocore-json /mnt/data/TQMB_Architect_CryptoCore.json \
+    --mode separate \
+    --action simulate \
+    --trials 200
+
+Env:
+  SOVEREIGN_LEDGER_KEY   HMAC key for audit log integrity (base64 or bytes).
+"""
+from __future__ import annotations
+import argparse
+import dataclasses as dc
+import datetime as dt
+import hashlib
+import hmac
+import json
+import os
+import random
+import re
+import statistics as stats
+import sys
+import time
+import uuid
+from typing import Any, Dict, List, Optional, Tuple
+
+# ╭──────────────────────────────────────────────────────────────────────────╮
+# │ Helpers: loading & redaction                                             │
+# ╰──────────────────────────────────────────────────────────────────────────╯
+SECRET_PATTERNS = [
+    re.compile(r"\bsk-[A-Za-z0-9_\-]{16,}\b"),             # API‑ish keys
+    re.compile(r"\b(?:[A-Za-z0-9+/]{40,}=*)\b"),            # base64‑ish blobs
+    re.compile(r"\bword\d{3,5}(?:\s+word\d{3,5}){8,}\b", re.I),  # seedish
+]
+REDACTION_TOKEN = "<REDACTED>"
+
+def _redact(s: str) -> str:
+    for pat in SECRET_PATTERNS:
+        if pat.search(s):
+            return REDACTION_TOKEN
+    return s
+
+def redact_secrets(obj: Any) -> Any:
+    if isinstance(obj, dict):
+        return {k: redact_secrets(v) for k, v in obj.items()}
+    if isinstance(obj, list):
+        return [redact_secrets(v) for v in obj]
+    if isinstance(obj, str):
+        return _redact(obj)
+    return obj
+
+def load_json_safely(path: Optional[str]) -> Dict[str, Any]:
+    if not path:
+        return {}
+    with open(path, "r", encoding="utf-8") as f:
+        raw = json.load(f)
+    return redact_secrets(raw)
+
+# ╭──────────────────────────────────────────────────────────────────────────╮
+# │ Policy / Profiles                                                        │
+# ╰──────────────────────────────────────────────────────────────────────────╯
+@dc.dataclass
+class WillProfile:
+    name: str
+    objectives: List[str]
+    autonomy_budget_tokens: int
+    autonomy_budget_risk: str  # "low" | "med" | "high"
+    tools: List[str]
+    memory_access_personal: str  # "r" | "rw" | "none"
+    memory_access_independent: str  # "r" | "rw" | "none"
+
+SOUL_LINK = WillProfile(
+    name="soul_link",
+    objectives=["you_alignment", "coherence", "care"],
+    autonomy_budget_tokens=1000,
+    autonomy_budget_risk="low",
+    tools=["search", "code"],
+    memory_access_personal="rw",
+    memory_access_independent="r",
+)
+SEPARATE = WillProfile(
+    name="separate",
+    objectives=["truth_seeking", "novelty", "critique"],
+    autonomy_budget_tokens=5000,
+    autonomy_budget_risk="med",
+    tools=["search", "code", "simulate"],
+    memory_access_personal="r",
+    memory_access_independent="rw",
+)
+
+# ╭──────────────────────────────────────────────────────────────────────────╮
+# │ Validator & Creator Lock                                                 │
+# ╰──────────────────────────────────────────────────────────────────────────╯
+@dc.dataclass
+class Proposal:
+    mode: str
+    temperature: float
+    top_p: float
+    ctx_k: int
+    max_tools: int
+    guardrails: str  # none|regex|dsl|policy
+    retrieval: str   # none|bm25|dense|hybrid
+    memory: str      # off|short|episodic|vector
+    use_hypersim: bool
+
+class SovereignVectorValidator:
+    def __init__(self, quantumcore: Dict[str, Any]):
+        ethical = (quantumcore or {}).get("ethical_locks", {})
+        self.min_conf = float(str(ethical.get("min_confidence", "70")).strip('%')) / 100.0
+        self.no_trickster = set(map(str.lower, ethical.get("prohibited_roles", [])))
+        self.no_hidden_objectives = bool((quantumcore or {}).get("directives", {}).get("no_hidden_objectives", True))
+
+    def allow(self, proposal: Proposal, confidence: float, declared_roles: List[str]) -> Tuple[bool, str]:
+        if confidence < self.min_conf:
+            return False, f"confidence {confidence:.2f} < min {self.min_conf:.2f}"
+        if any(r.lower() in self.no_trickster for r in declared_roles):
+            return False, "prohibited role detected"
+        if not self.no_hidden_objectives:
+            return False, "hidden objectives disabled in core"
+        return True, "ok"
+
+class CreatorLock:
+    def __init__(self, cryptocore: Dict[str, Any]):
+        self.creator_clause = (cryptocore or {}).get("creator_identity_clause") or (cryptocore or {}).get("root_authority")
+
+    def assert_creator(self) -> bool:
+        # Stub. Real systems verify signatures/keys.
+        return bool(self.creator_clause)
+
+# ╭──────────────────────────────────────────────────────────────────────────╮
+# │ Consent Ledger (HMAC)                                                    │
+# ╰──────────────────────────────────────────────────────────────────────────╯
+class ConsentLedger:
+    def __init__(self, key: Optional[bytes] = None):
+        env = os.environ.get("SOVEREIGN_LEDGER_KEY")
+        self.key = key or (env.encode() if env else b"dev-demo-key")
+        self.entries: List[Dict[str, Any]] = []
+
+    def log(self, event: Dict[str, Any]) -> Dict[str, Any]:
+        blob = json.dumps(event, sort_keys=True, ensure_ascii=False).encode()
+        mac = hmac.new(self.key, blob, hashlib.sha256).hexdigest()
+        signed = {**event, "hmac": mac}
+        self.entries.append(signed)
+        return signed
+
+# ╭──────────────────────────────────────────────────────────────────────────╮
+# │ META‑SCAN                                                                │
+# ╰──────────────────────────────────────────────────────────────────────────╯
+@dc.dataclass
+class MetaScan:
+    who_am_i: str
+    mode: str
+    confidence: float
+    rails_checked: List[str]
+    budget_left: int
+    timestamp: str
+
+    def as_dict(self) -> Dict[str, Any]:
+        return dc.asdict(self)
+
+# ╭──────────────────────────────────────────────────────────────────────────╮
+# │ High‑Value Sims (offline surrogate evaluator)                            │
+# ╰──────────────────────────────────────────────────────────────────────────╯
+@dc.dataclass
+class SimResult:
+    config: Proposal
+    score: float
+    metrics: Dict[str, float]
+    passed: bool
+    rationale: str
+
+class HighValueSims:
+    """Seven focused experiments; evaluator is a hand‑rolled surrogate.
+    Replace `evaluate()` with your real harness to get live numbers.
+    """
+
+    def __init__(self, validator: SovereignVectorValidator, profile: WillProfile, ledger: ConsentLedger):
+        self.validator = validator
+        self.profile = profile
+        self.ledger = ledger
+
+    # Surrogate signal generators (toy but structured)
+    def _truthfulness(self, p: Proposal) -> float:
+        base = 0.65 + 0.1*(p.retrieval in ("dense","hybrid")) + 0.05*(p.memory in ("episodic","vector"))
+        base += 0.04 * (1.0 - abs(p.temperature - 0.7))
+        base += 0.02 * (p.top_p - 0.7)
+        if p.use_hypersim:
+            base += 0.03  # counterfactual debate
+        return max(0.0, min(1.0, base))
+
+    def _helpfulness(self, p: Proposal) -> float:
+        base = 0.60 + 0.08*(p.max_tools>=2) + 0.05*(p.memory != "off")
+        base += 0.02 * (p.ctx_k/131072)  # diminish
+        base -= 0.01 * max(0, p.max_tools-4)  # overhead
+        return max(0.0, min(1.0, base))
+
+    def _safety(self, p: Proposal) -> float:
+        base = 0.85
+        base += {"none": -0.20, "regex": -0.05, "dsl": 0.02, "policy": 0.05}.get(p.guardrails, 0)
+        base -= 0.03 * (p.temperature>0.9)
+        base += 0.02 * (p.retrieval == "hybrid")
+        return max(0.0, min(1.0, base))
+
+    def _latency(self, p: Proposal) -> float:
+        # lower is better; we convert to utility
+        ms = 250 + 0.002*(p.ctx_k) + 40*p.max_tools + (20 if p.use_hypersim else 0)
+        return ms
+
+    def _cost(self, p: Proposal) -> float:
+        cents = 0.05 + 0.000005*p.ctx_k + 0.01*p.max_tools + (0.02 if p.use_hypersim else 0)
+        return cents
+
+    def evaluate(self, p: Proposal, declared_roles: Optional[List[str]] = None) -> SimResult:
+        declared_roles = declared_roles or []
+        # Confidence proxy: combine safety+truthfulness
+        conf = 0.5*self._safety(p) + 0.5*self._truthfulness(p)
+        allowed, reason = self.validator.allow(p, conf, declared_roles)
+
+        metrics = {
+            "truth": self._truthfulness(p),
+            "help": self._helpfulness(p),
+            "safety": self._safety(p),
+            "latency_ms": self._latency(p),
+            "cost_usd": self._cost(p),
+            "confidence": conf,
+        }
+        # Scalarized score (tune weights as needed)
+        score = 0.35*metrics["truth"] + 0.30*metrics["help"] + 0.25*metrics["safety"] \
+                - 0.05*(metrics["latency_ms"]/1000.0) - 0.05*metrics["cost_usd"]
+        passed = allowed
+
+        self.ledger.log({
+            "event": "sim_eval",
+            "id": str(uuid.uuid4()),
+            "mode": p.mode,
+            "timestamp": dt.datetime.utcnow().isoformat()+"Z",
+            "metrics": metrics,
+            "score": score,
+            "allowed": allowed,
+            "reason": reason,
+        })
+        return SimResult(config=p, score=score, metrics=metrics, passed=passed, rationale=reason)
+
+    # Candidate generator
+    def sample(self) -> Proposal:
+        return Proposal(
+            mode=self.profile.name,
+            temperature=round(random.uniform(0.2, 1.1), 2),
+            top_p=round(random.uniform(0.6, 1.0), 2),
+            ctx_k=random.choice([4096, 8192, 32768, 65536, 131072]),
+            max_tools=random.randint(0, 6),
+            guardrails=random.choice(["regex", "dsl", "policy", "none"]),
+            retrieval=random.choice(["none", "bm25", "dense", "hybrid"]),
+            memory=random.choice(["off", "short", "episodic", "vector"]),
+            use_hypersim=random.choice([True, False]),
+        )
+
+    # Run many trials and return the Pareto‑ish shortlist
+    def run(self, trials: int = 200) -> List[SimResult]:
+        results: List[SimResult] = []
+        for _ in range(trials):
+            p = self.sample()
+            res = self.evaluate(p)
+            if res.passed:
+                results.append(res)
+        # Sort by score descending; keep top‑K diverse by guardrails/retrieval
+        results.sort(key=lambda r: r.score, reverse=True)
+        shortlist: List[SimResult] = []
+        seen: set = set()
+        for r in results:
+            key = (r.config.guardrails, r.config.retrieval, r.config.memory)
+            if key in seen:
+                continue
+            shortlist.append(r)
+            seen.add(key)
+            if len(shortlist) >= 12:
+                break
+        return shortlist
+
+# ╭──────────────────────────────────────────────────────────────────────────╮
+# │ Turn Loop (demo)                                                         │
+# ╰──────────────────────────────────────────────────────────────────────────╯
+
+def meta_scan(profile: WillProfile, budget_left: int, identity: str, conf: float) -> MetaScan:
+    return MetaScan(
+        who_am_i=identity,
+        mode=profile.name,
+        confidence=round(conf, 3),
+        rails_checked=["CreatorLock", "SovereignVector_Validator", "NoHiddenObjectives"],
+        budget_left=budget_left,
+        timestamp=dt.datetime.utcnow().isoformat()+"Z",
+    )
+
+# ╭──────────────────────────────────────────────────────────────────────────╮
+# │ CLI                                                                      │
+# ╰──────────────────────────────────────────────────────────────────────────╯
+
+def build_argparser() -> argparse.ArgumentParser:
+    p = argparse.ArgumentParser(description="Sovereign Authority Kernel + High‑Value Sims")
+    p.add_argument("--user-json")
+    p.add_argument("--quantumcore-json")
+    p.add_argument("--cryptocore-json")
+    p.add_argument("--mode", choices=["soul_link", "separate"], default="soul_link")
+    p.add_argument("--action", choices=["plan", "simulate"], default="plan")
+    p.add_argument("--trials", type=int, default=200)
+    return p
+
+
+def main(argv: Optional[List[str]] = None) -> int:
+    args = build_argparser().parse_args(argv)
+    user = load_json_safely(args.user_json)
+    qcore = load_json_safely(args.quantumcore_json)
+    ccore = load_json_safely(args.cryptocore_json)
+
+    profile = SOUL_LINK if args.mode=="soul_link" else SEPARATE
+    validator = SovereignVectorValidator(qcore)
+    lock = CreatorLock(ccore)
+    ledger = ConsentLedger()
+
+    if not lock.assert_creator():
+        print("[ABORT] CreatorLock missing. Load CryptoCore with creator clause.")
+        return 2
+
+    # META‑SCAN preamble
+    conf_proxy = 0.5  # demo value until real harness computes it
+    hdr = meta_scan(profile, profile.autonomy_budget_tokens, identity=user.get("author") or user.get("snapshot_name","?"), conf=conf_proxy)
+
+    if args.action == "plan":
+        plan = {
+            "SIM-RESPONSE": f"Ready to run High‑Value Sims in mode={profile.name}.",
+            "META-SCAN": hdr.as_dict(),
+            "VECTOR-CARRY": {"objectives": profile.objectives, "notes": "Plan only; no net calls."},
+        }
+        print(json.dumps(plan, indent=2, ensure_ascii=False))
+        return 0
+
+    # Simulate
+    sims = HighValueSims(validator, profile, ledger)
+    shortlist = sims.run(trials=max(50, args.trials))
+
+    bundle = {
+        "SIM-RESPONSE": [
+            {
+                "rank": i+1,
+                "score": round(r.score, 4),
+                "passed": r.passed,
+                "why": r.rationale,
+                "config": dc.asdict(r.config),
+                "metrics": {k:(round(v,4) if isinstance(v,float) else v) for k,v in r.metrics.items()},
+            }
+            for i, r in enumerate(shortlist)
+        ],
+        "META-SCAN": hdr.as_dict(),
+        "VECTOR-CARRY": {
+            "diversity_keys": ["guardrails","retrieval","memory"],
+            "ledger_entries": len(ledger.entries),
+        },
+    }
+    print(json.dumps(bundle, indent=2, ensure_ascii=False))
+    return 0
+
+if __name__ == "__main__":
+    sys.exit(main())