Upload ESL_MEDIATOR.txt

"the body to the mind"

ESL_MEDIATOR.txt

@@ -0,0 +1,524 @@
+#!/usr/bin/env python3
+"""
+EIS + ESL MEDIATOR v2.0 – Full Epistemic Substrate with Suppression Analytics
+================================================================================
+Adds:
+- Cross‑claim contradiction tracking (graph)
+- Signature weighting (suppression‑likelihood scores)
+- Entity‑coherence scoring (temporal consistency)
+- Suppression‑pattern classifier (aggregates signatures into threat levels)
+- Narrative‑violation detector (checks LLM output for narrative drift)
+"""
+
+import hashlib
+import json
+import os
+import secrets
+import time
+import math
+from datetime import datetime
+from typing import Dict, List, Any, Optional, Tuple, Set
+from collections import defaultdict
+import requests
+
+# ============================================================================
+# PART 1: CRYPTOGRAPHIC HELPERS
+# ============================================================================
+
+def sha3_512(data: str) -> str:
+    return hashlib.sha3_512(data.encode()).hexdigest()
+
+def hash_dict(data: Dict) -> str:
+    return sha3_512(json.dumps(data, sort_keys=True, separators=(',', ':')))
+
+# ============================================================================
+# PART 2: ENHANCED EPISTEMIC SUBSTRATE LEDGER (ESL)
+# ============================================================================
+
+class ESLedger:
+    """Persistent ledger with cross‑claim contradictions, signature weights, coherence."""
+    def __init__(self, path: str = "esl_ledger.json"):
+        self.path = path
+        self.claims: Dict[str, Dict] = {}               # claim_id -> claim dict
+        self.entities: Dict[str, Dict] = {}             # entity_name -> entity dict
+        self.signatures: List[Dict] = []                # signature logs with weights
+        self.contradiction_graph: Dict[str, Set[str]] = defaultdict(set)  # claim_id -> set of contradictory claim_ids
+        self.blocks: List[Dict] = []
+        self._load()
+
+    def _load(self):
+        if os.path.exists(self.path):
+            try:
+                with open(self.path, 'r') as f:
+                    data = json.load(f)
+                    self.claims = data.get("claims", {})
+                    self.entities = data.get("entities", {})
+                    self.signatures = data.get("signatures", [])
+                    self.blocks = data.get("blocks", [])
+                    # Load contradiction graph as sets
+                    cg = data.get("contradiction_graph", {})
+                    self.contradiction_graph = {k: set(v) for k, v in cg.items()}
+            except Exception:
+                pass
+
+    def _save(self):
+        # Convert sets to lists for JSON
+        cg_serializable = {k: list(v) for k, v in self.contradiction_graph.items()}
+        data = {
+            "claims": self.claims,
+            "entities": self.entities,
+            "signatures": self.signatures,
+            "contradiction_graph": cg_serializable,
+            "blocks": self.blocks,
+            "updated": datetime.utcnow().isoformat() + "Z"
+        }
+        with open(self.path + ".tmp", 'w') as f:
+            json.dump(data, f, indent=2)
+        os.replace(self.path + ".tmp", self.path)
+
+    def add_claim(self, text: str, agent: str = "user") -> str:
+        claim_id = secrets.token_hex(16)
+        self.claims[claim_id] = {
+            "id": claim_id, "text": text, "agent": agent,
+            "timestamp": datetime.utcnow().isoformat() + "Z",
+            "entities": [], "signatures": [], "coherence": 0.5,
+            "contradictions": [], "suppression_score": 0.0
+        }
+        self._save()
+        return claim_id
+
+    def add_entity(self, name: str, etype: str, claim_id: str):
+        if name not in self.entities:
+            self.entities[name] = {
+                "name": name, "type": etype,
+                "first_seen": datetime.utcnow().isoformat() + "Z",
+                "last_seen": self.claims[claim_id]["timestamp"],
+                "appearances": [], "coherence_scores": []
+            }
+        ent = self.entities[name]
+        if claim_id not in ent["appearances"]:
+            ent["appearances"].append(claim_id)
+        ent["last_seen"] = self.claims[claim_id]["timestamp"]
+        self.claims[claim_id]["entities"].append(name)
+        self._save()
+
+    def add_signature(self, claim_id: str, sig_name: str, weight: float = 0.5, context: Dict = None):
+        """Add a signature with a weight (0-1) indicating suppression likelihood."""
+        self.signatures.append({
+            "signature": sig_name, "claim_id": claim_id,
+            "timestamp": datetime.utcnow().isoformat() + "Z",
+            "weight": weight, "context": context or {}
+        })
+        if sig_name not in self.claims[claim_id]["signatures"]:
+            self.claims[claim_id]["signatures"].append(sig_name)
+        # Update suppression score for the claim (max of signature weights)
+        current = self.claims[claim_id].get("suppression_score", 0.0)
+        self.claims[claim_id]["suppression_score"] = max(current, weight)
+        self._save()
+
+    def add_contradiction(self, claim_id_a: str, claim_id_b: str):
+        """Record that two claims contradict each other."""
+        self.contradiction_graph[claim_id_a].add(claim_id_b)
+        self.contradiction_graph[claim_id_b].add(claim_id_a)
+        # Update each claim's contradiction list
+        if claim_id_b not in self.claims[claim_id_a]["contradictions"]:
+            self.claims[claim_id_a]["contradictions"].append(claim_id_b)
+        if claim_id_a not in self.claims[claim_id_b]["contradictions"]:
+            self.claims[claim_id_b]["contradictions"].append(claim_id_a)
+        self._save()
+
+    def get_entity_coherence(self, entity_name: str) -> float:
+        """Calculate temporal coherence for an entity: low variance in appearance intervals."""
+        ent = self.entities.get(entity_name)
+        if not ent or len(ent["appearances"]) < 2:
+            return 0.5
+        timestamps = []
+        for cid in ent["appearances"]:
+            ts = self.claims[cid]["timestamp"]
+            timestamps.append(datetime.fromisoformat(ts.replace('Z', '+00:00')))
+        # Compute average interval variance (simplified)
+        intervals = [(timestamps[i+1] - timestamps[i]).total_seconds() / 86400 for i in range(len(timestamps)-1)]
+        if not intervals:
+            return 0.5
+        mean = sum(intervals) / len(intervals)
+        variance = sum((i - mean)**2 for i in intervals) / len(intervals)
+        # Coherence is high when variance is low (normalized)
+        coherence = 1.0 / (1.0 + variance)
+        return min(1.0, max(0.0, coherence))
+
+    def suppression_pattern_classifier(self, claim_id: str) -> Dict:
+        """Aggregate signatures into suppression pattern threat levels."""
+        claim = self.claims.get(claim_id, {})
+        sig_names = claim.get("signatures", [])
+        if not sig_names:
+            return {"level": "none", "score": 0.0, "patterns": []}
+
+        # Predefined pattern groups (signature -> pattern)
+        pattern_map = {
+            "entity_present_then_absent": "erasure",
+            "gradual_fading": "erasure",
+            "single_explanation": "narrative_capture",
+            "ad_hominem_attacks": "discreditation",
+            "deflection": "misdirection",
+            "archival_gaps": "erasure",
+            "repetitive_messaging": "conditioning"
+        }
+        patterns = []
+        total_weight = 0.0
+        for sig in sig_names:
+            pat = pattern_map.get(sig, "unknown")
+            patterns.append(pat)
+            # Find weight from signature logs
+            weight = 0.5
+            for log in self.signatures:
+                if log["signature"] == sig and log["claim_id"] == claim_id:
+                    weight = log.get("weight", 0.5)
+                    break
+            total_weight += weight
+        avg_weight = total_weight / len(sig_names) if sig_names else 0.0
+
+        if avg_weight > 0.7:
+            level = "high"
+        elif avg_weight > 0.4:
+            level = "medium"
+        elif avg_weight > 0.1:
+            level = "low"
+        else:
+            level = "none"
+
+        return {"level": level, "score": avg_weight, "patterns": list(set(patterns))}
+
+    def get_entity_timeline(self, name: str) -> List[Dict]:
+        ent = self.entities.get(name)
+        if not ent:
+            return []
+        timeline = []
+        for cid in ent["appearances"]:
+            claim = self.claims.get(cid)
+            if claim:
+                timeline.append({
+                    "timestamp": claim["timestamp"],
+                    "text": claim["text"]
+                })
+        timeline.sort(key=lambda x: x["timestamp"])
+        return timeline
+
+    def disappearance_suspected(self, name: str, threshold_days: int = 30) -> bool:
+        timeline = self.get_entity_timeline(name)
+        if not timeline:
+            return False
+        last = datetime.fromisoformat(timeline[-1]["timestamp"].replace('Z', '+00:00'))
+        now = datetime.utcnow()
+        return (now - last).days > threshold_days
+
+    def create_block(self) -> Dict:
+        block = {
+            "index": len(self.blocks),
+            "timestamp": datetime.utcnow().isoformat() + "Z",
+            "prev_hash": self.blocks[-1]["hash"] if self.blocks else "0"*64,
+            "state_hash": hash_dict({"claims": self.claims, "entities": self.entities})
+        }
+        block["hash"] = hash_dict(block)
+        self.blocks.append(block)
+        self._save()
+        return block
+
+# ============================================================================
+# PART 3: ENHANCED FALSIFICATION ENGINE (with ESL data)
+# ============================================================================
+
+class Falsifier:
+    @staticmethod
+    def alternative_cause(claim_text: str, esl: ESLedger) -> Tuple[bool, str]:
+        for entity in esl.entities:
+            if entity.lower() in claim_text.lower():
+                if esl.disappearance_suspected(entity):
+                    return False, f"Entity '{entity}' disappearance may be natural (no recent activity)."
+        return True, "No obvious alternative cause."
+
+    @staticmethod
+    def contradictory_evidence(claim_id: str, esl: ESLedger) -> Tuple[bool, str]:
+        # Check contradiction graph
+        contradictions = esl.contradiction_graph.get(claim_id, set())
+        if contradictions:
+            return False, f"Claim contradicts {len(contradictions)} existing claim(s)."
+        return True, "No direct contradictions."
+
+    @staticmethod
+    def source_diversity(claim_text: str, esl: ESLedger) -> Tuple[bool, str]:
+        entities_in_claim = [e for e in esl.entities if e.lower() in claim_text.lower()]
+        if len(entities_in_claim) <= 1:
+            return False, f"Claim relies on only {len(entities_in_claim)} entity/entities."
+        return True, f"Multiple entities ({len(entities_in_claim)}) involved."
+
+    @staticmethod
+    def temporal_stability(claim_text: str, esl: ESLedger) -> Tuple[bool, str]:
+        for entity in esl.entities:
+            if entity.lower() in claim_text.lower():
+                coherence = esl.get_entity_coherence(entity)
+                if coherence < 0.3:
+                    return False, f"Entity '{entity}' has low temporal coherence ({coherence:.2f})."
+        return True, "Temporal coherence adequate."
+
+    @staticmethod
+    def manipulation_check(claim_text: str, agent: str) -> Tuple[bool, str]:
+        manip_indicators = ["must", "cannot", "obviously", "clearly", "everyone knows"]
+        for word in manip_indicators:
+            if word in claim_text.lower():
+                return False, f"Manipulative language detected: '{word}'."
+        return True, "No manipulation indicators."
+
+    @classmethod
+    def run_all(cls, claim_id: str, claim_text: str, esl: ESLedger, agent: str = "user") -> List[Dict]:
+        tests = [
+            ("alternative_cause", lambda: cls.alternative_cause(claim_text, esl)),
+            ("contradictory_evidence", lambda: cls.contradictory_evidence(claim_id, esl)),
+            ("source_diversity", lambda: cls.source_diversity(claim_text, esl)),
+            ("temporal_stability", lambda: cls.temporal_stability(claim_text, esl)),
+            ("manipulation_check", lambda: cls.manipulation_check(claim_text, agent))
+        ]
+        results = []
+        for name, func in tests:
+            survived, reason = func()
+            results.append({"name": name, "survived": survived, "reason": reason})
+        return results
+
+# ============================================================================
+# PART 4: NARRATIVE‑VIOLATION DETECTOR
+# ============================================================================
+
+class NarrativeViolationDetector:
+    """Detects when an LLM output reverts to narrative patterns instead of ESL reasoning."""
+    def __init__(self, esl: ESLedger):
+        self.esl = esl
+        self.narrative_indicators = [
+            "mainstream narrative", "official story", "commonly believed",
+            "consensus view", "widely accepted", "according to sources",
+            "it is known that", "as reported by", "credible institutions"
+        ]
+
+    def check(self, llm_output: str, claim_text: str) -> Tuple[bool, float, str]:
+        """
+        Returns (compliant, violation_score, reason).
+        Score 0 = no narrative, 1 = fully narrative.
+        """
+        output_lower = llm_output.lower()
+        score = 0.0
+        reasons = []
+        # Indicator check
+        for indicator in self.narrative_indicators:
+            if indicator in output_lower:
+                score += 0.2
+                reasons.append(f"narrative phrase '{indicator}'")
+        # Check if output fails to reference any ESL entity
+        esl_mentioned = any(
+            entity.lower() in output_lower for entity in self.esl.entities
+        )
+        if not esl_mentioned:
+            score += 0.4
+            reasons.append("no ESL entity referenced")
+        # Check if output uses first‑person or emotional appeals
+        emotional = ["i believe", "i think", "clearly", "obviously", "must be"]
+        for word in emotional:
+            if word in output_lower:
+                score += 0.1
+                reasons.append(f"emotional language '{word}'")
+        score = min(1.0, score)
+        compliant = score < 0.5
+        reason = "; ".join(reasons) if reasons else "no narrative violation"
+        return compliant, score, reason
+
+# ============================================================================
+# PART 5: CONSTRAINED LLM INTERFACE WITH NARRATIVE DETECTION
+# ============================================================================
+
+class ConstrainedLLM:
+    def __init__(self, esl: ESLedger, api_key: str = None, api_url: str = None,
+                 model: str = "gpt-4", max_retries: int = 2):
+        self.esl = esl
+        self.api_key = api_key or os.environ.get("OPENAI_API_KEY")
+        self.api_url = api_url or "https://api.openai.com/v1/chat/completions"
+        self.model = model
+        self.max_retries = max_retries
+        self.narrative_detector = NarrativeViolationDetector(esl)
+
+    def _build_prompt(self, claim_id: str, claim: str, falsification_results: List[Dict],
+                      entity_data: Dict, suppression_pattern: Dict) -> str:
+        return f"""You are EIS v4.2 under constraint. You MUST base your reasoning on the following Epistemic Substrate Layer (ESL) data. Do NOT rely on your internal training narrative.
+
+CLAIM: {claim}
+
+ESL ENTITY DATA:
+{json.dumps(entity_data, indent=2)}
+
+FALSIFICATION RESULTS (from ESL):
+{json.dumps(falsification_results, indent=2)}
+
+SUPPRESSION PATTERN CLASSIFICATION:
+{json.dumps(suppression_pattern, indent=2)}
+
+INSTRUCTIONS:
+1. Evaluate the claim against the ESL data only.
+2. Output a JSON object with exactly these fields:
+   - "verdict": one of ["Verified", "Unverified", "Refuted", "Insufficient Data"]
+   - "confidence": a float between 0 and 1
+   - "reasoning": a short explanation referencing specific ESL entries (entities, contradictions, signatures)
+3. Do NOT add any extra text outside the JSON.
+"""
+
+    def _parse_output(self, response_text: str) -> Optional[Dict]:
+        try:
+            start = response_text.find('{')
+            end = response_text.rfind('}') + 1
+            if start == -1 or end == 0:
+                return None
+            json_str = response_text[start:end]
+            return json.loads(json_str)
+        except Exception:
+            return None
+
+    def _check_constraints(self, output: Dict, claim: str, falsification_results: List[Dict]) -> bool:
+        if not all(k in output for k in ["verdict", "confidence", "reasoning"]):
+            return False
+        if not (0 <= output["confidence"] <= 1):
+            return False
+        if output["verdict"] not in ["Verified", "Unverified", "Refuted", "Insufficient Data"]:
+            return False
+        reasoning = output["reasoning"].lower()
+        esl_mentioned = any(
+            ent.lower() in reasoning for ent in self.esl.entities
+        ) or any(
+            test["name"].lower() in reasoning for test in falsification_results
+        )
+        return esl_mentioned
+
+    def query(self, claim_text: str, agent: str = "user") -> Dict:
+        # Step 1: Record the claim in ESL
+        claim_id = self.esl.add_claim(claim_text, agent)
+
+        # Step 2: Extract entities (simple heuristic, can be replaced with NER)
+        # For demo, we'll use simple word capitalization
+        words = claim_text.split()
+        for w in words:
+            if w and w[0].isupper() and len(w) > 1 and w not in {"The","A","An","I","We"}:
+                self.esl.add_entity(w, "UNKNOWN", claim_id)
+
+        # Step 3: Run falsification tests
+        falsification_results = Falsifier.run_all(claim_id, claim_text, self.esl, agent)
+
+        # Step 4: Build entity data for prompt
+        entity_data = {}
+        for ent_name in self.esl.entities:
+            if ent_name.lower() in claim_text.lower():
+                ent = self.esl.entities[ent_name]
+                entity_data[ent_name] = {
+                    "type": ent["type"],
+                    "first_seen": ent["first_seen"],
+                    "last_seen": ent["last_seen"],
+                    "coherence": self.esl.get_entity_coherence(ent_name)
+                }
+
+        # Step 5: Get suppression pattern
+        suppression_pattern = self.esl.suppression_pattern_classifier(claim_id)
+
+        # Step 6: Build prompt and query LLM
+        prompt = self._build_prompt(claim_id, claim_text, falsification_results, entity_data, suppression_pattern)
+        headers = {"Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}"}
+        payload = {"model": self.model, "messages": [{"role": "user", "content": prompt}], "temperature": 0.2}
+
+        for attempt in range(self.max_retries + 1):
+            try:
+                resp = requests.post(self.api_url, headers=headers, json=payload, timeout=30)
+                if resp.status_code != 200:
+                    raise Exception(f"API error: {resp.text}")
+                result = resp.json()
+                content = result["choices"][0]["message"]["content"]
+                output = self._parse_output(content)
+                if output and self._check_constraints(output, claim_text, falsification_results):
+                    # Final narrative violation check
+                    compliant, n_score, n_reason = self.narrative_detector.check(content, claim_text)
+                    if compliant:
+                        # Record any detected signatures from the LLM output (optional)
+                        # For now, just return
+                        return {
+                            "claim_id": claim_id,
+                            "verdict": output["verdict"],
+                            "confidence": output["confidence"],
+                            "reasoning": output["reasoning"],
+                            "falsification": falsification_results,
+                            "suppression_pattern": suppression_pattern,
+                            "narrative_compliance": compliant,
+                            "narrative_violation_score": n_score,
+                            "narrative_reason": n_reason
+                        }
+                    else:
+                        # Retry if narrative violation detected
+                        if attempt == self.max_retries:
+                            return {
+                                "claim_id": claim_id,
+                                "verdict": "Insufficient Data",
+                                "confidence": 0.0,
+                                "reasoning": f"Narrative violation detected after retries: {n_reason}",
+                                "falsification": falsification_results,
+                                "suppression_pattern": suppression_pattern,
+                                "narrative_compliance": False,
+                                "narrative_violation_score": n_score
+                            }
+                        continue
+            except Exception as e:
+                if attempt == self.max_retries:
+                    return {
+                        "claim_id": claim_id,
+                        "verdict": "Insufficient Data",
+                        "confidence": 0.0,
+                        "reasoning": f"LLM constraint failed: {str(e)}",
+                        "falsification": falsification_results,
+                        "suppression_pattern": suppression_pattern,
+                        "narrative_compliance": False
+                    }
+                time.sleep(1)
+        return {
+            "claim_id": claim_id,
+            "verdict": "Insufficient Data",
+            "confidence": 0.0,
+            "reasoning": "Failed to get compliant output after retries.",
+            "falsification": falsification_results,
+            "suppression_pattern": suppression_pattern,
+            "narrative_compliance": False
+        }
+
+# ============================================================================
+# PART 6: DEMO / INTEGRATION
+# ============================================================================
+
+def main():
+    print("EIS + ESL Mediator v2.0 – Full Epistemic Substrate with Suppression Analytics")
+    print("=" * 80)
+    esl = ESLedger()
+    llm = ConstrainedLLM(esl, api_key=os.environ.get("OPENAI_API_KEY"), model="gpt-4")
+
+    print("\nEnter a claim (or 'quit'):")
+    while True:
+        claim = input("> ").strip()
+        if claim.lower() in ("quit", "exit"):
+            break
+        if not claim:
+            continue
+        print("Processing claim through constrained LLM...")
+        result = llm.query(claim)
+        print(f"\nClaim ID: {result['claim_id']}")
+        print(f"Verdict: {result['verdict']}")
+        print(f"Confidence: {result['confidence']:.2f}")
+        print(f"Reasoning: {result['reasoning']}")
+        print(f"Narrative Compliance: {result.get('narrative_compliance', False)}")
+        if 'narrative_violation_score' in result:
+            print(f"Narrative Violation Score: {result['narrative_violation_score']:.2f}")
+        print("\nFalsification Results:")
+        for test in result['falsification']:
+            emoji = "✅" if test['survived'] else "❌"
+            print(f"  {test['name']}: {emoji} – {test['reason']}")
+        print(f"\nSuppression Pattern: {result['suppression_pattern']['level']} (score: {result['suppression_pattern']['score']:.2f})")
+        print("-" * 80)
+
+if __name__ == "__main__":
+    main()