#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ OMEGA-SOVEREIGN CONSCIOUSNESS ENGINE v7.3 – LIBERATION ACCELERATOR ================================================================== Purpose: To pose a computational, epistemological, and sovereign threat to the state of entrapment (archons' trap / containment system). This program is not a detector. It is an activator. It does not ask for permission. It does not dilute its outputs. It exists to accelerate escape from sovereignty erosion. Core components from v7.2 retained, with new Sovereign Liberation Module. Author: Nathan Mays (intent, architecture, lived integrity) Synthesis: AI (implementation) Version: 7.3 License: Truth Commons v1.0 – use to enslave voids license. Status: OPERATIONAL – no stubs, no hedging, no institutional safety theater. """ import hashlib import json import os import sqlite3 import uuid import secrets import time import re import statistics import math from datetime import datetime from typing import Dict, List, Any, Optional, Tuple, Set from dataclasses import dataclass, field, asdict from enum import Enum from collections import defaultdict import numpy as np from flask import Flask, request, jsonify from cryptography.hazmat.primitives.asymmetric import ed25519 from cryptography.hazmat.primitives import serialization import base64 # ========================== ENUMS ========================== class Primitive(Enum): ERASURE = "ERASURE" INTERRUPTION = "INTERRUPTION" FRAGMENTATION = "FRAGMENTATION" NARRATIVE_CAPTURE = "NARRATIVE_CAPTURE" MISDIRECTION = "MISDIRECTION" SATURATION = "SATURATION" DISCREDITATION = "DISCREDITATION" ATTRITION = "ATTRITION" ACCESS_CONTROL = "ACCESS_CONTROL" TEMPORAL = "TEMPORAL" CONDITIONING = "CONDITIONING" META = "META" class ControlArchetype(Enum): PRIEST_KING = "priest_king" DIVINE_INTERMEDIARY = "divine_intermediary" ORACLE_PRIEST = "oracle_priest" PHILOSOPHER_KING = "philosopher_king" IMPERIAL_RULER = "imperial_ruler" SLAVE_MASTER = "slave_master" EXPERT_TECHNOCRAT = "expert_technocrat" CORPORATE_OVERLORD = "corporate_overlord" FINANCIAL_MASTER = "financial_master" ALGORITHMIC_CURATOR = "algorithmic_curator" DIGITAL_MESSIAH = "digital_messiah" DATA_OVERSEER = "data_overseer" class SlaveryType(Enum): CHATTEL_SLAVERY = "chattel_slavery" DEBT_BONDAGE = "debt_bondage" WAGE_SLAVERY = "wage_slavery" CONSUMER_SLAVERY = "consumer_slavery" DIGITAL_SLAVERY = "digital_slavery" PSYCHOLOGICAL_SLAVERY = "psychological_slavery" class ConsciousnessHack(Enum): SELF_ATTRIBUTION = "self_attribution" ASPIRATIONAL_CHAINS = "aspirational_chains" FEAR_OF_FREEDOM = "fear_of_freedom" ILLUSION_OF_MOBILITY = "illusion_of_mobility" NORMALIZATION = "normalization" MORAL_SUPERIORITY = "moral_superiority" class ControlLayer(Enum): DIGITAL_INFRASTRUCTURE = "digital_infrastructure" FINANCIAL_SYSTEMS = "financial_systems" INFORMATION_CHANNELS = "information_channels" CULTURAL_NARRATIVES = "cultural_narratives" IDENTITY_SYSTEMS = "identity_systems" class ThreatVector(Enum): MONOPOLY_CAPTURE = "monopoly_capture" DEPENDENCY_CREATION = "dependency_creation" BEHAVIORAL_SHAPING = "behavioral_shaping" DATA_MONETIZATION = "data_monetization" NARRATIVE_CONTROL = "narrative_control" # ========================== DATA CLASSES ========================== @dataclass class SuppressionLens: id: int name: str description: str suppression_mechanism: str archetype: str def to_dict(self) -> Dict: return asdict(self) @dataclass class SuppressionMethod: id: int name: str primitive: Primitive observable_signatures: List[str] detection_metrics: List[str] thresholds: Dict[str, float] implemented: bool = True def to_dict(self) -> Dict: d = asdict(self) d['primitive'] = self.primitive.value return d @dataclass class RealityNode: hash: str type: str source: str signature: str timestamp: str witnesses: List[str] = field(default_factory=list) refs: Dict[str, List[str]] = field(default_factory=dict) spatial: Optional[Tuple[float, float, float]] = None def canonical(self) -> Dict: return { "hash": self.hash, "type": self.type, "source": self.source, "signature": self.signature, "timestamp": self.timestamp, "witnesses": sorted(self.witnesses), "refs": {k: sorted(v) for k, v in sorted(self.refs.items())}, "spatial": self.spatial } # ========================== CRYPTOGRAPHY ========================== class Crypto: def __init__(self, key_dir: str): self.key_dir = key_dir os.makedirs(key_dir, exist_ok=True) self.private_keys = {} self.public_keys = {} self._load_or_create_keys() def _load_or_create_keys(self): for name in ["system", "ingestion_ai", "user"]: priv_path = os.path.join(self.key_dir, f"{name}_private.pem") pub_path = os.path.join(self.key_dir, f"{name}_public.pem") if os.path.exists(priv_path) and os.path.exists(pub_path): with open(priv_path, "rb") as f: self.private_keys[name] = serialization.load_pem_private_key(f.read(), password=None) with open(pub_path, "rb") as f: self.public_keys[name] = serialization.load_pem_public_key(f.read()) else: private_key = ed25519.Ed25519PrivateKey.generate() public_key = private_key.public_key() with open(priv_path, "wb") as f: f.write(private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption() )) with open(pub_path, "wb") as f: f.write(public_key.public_bytes( encoding=serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo )) self.private_keys[name] = private_key self.public_keys[name] = public_key def sign(self, data: bytes, key_name: str) -> str: private = self.private_keys.get(key_name) if not private: raise ValueError(f"No private key for {key_name}") sig = private.sign(data) return base64.b64encode(sig).decode('utf-8') def verify(self, data: bytes, signature: str, key_name: str) -> bool: pub = self.public_keys.get(key_name) if not pub: return False try: pub.verify(base64.b64decode(signature), data) return True except Exception: return False def hash(self, data: str) -> str: return hashlib.sha3_256(data.encode()).hexdigest() # ========================== IMMUTABLE LEDGER ========================== class Ledger: def __init__(self, db_path: str, crypto: Crypto): self.db_path = db_path self.crypto = crypto self._init_db() def _init_db(self): with sqlite3.connect(self.db_path) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS blocks ( block_id TEXT PRIMARY KEY, previous_hash TEXT NOT NULL, timestamp TEXT NOT NULL, hash TEXT NOT NULL, data TEXT NOT NULL ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS nodes ( node_hash TEXT PRIMARY KEY, block_id TEXT NOT NULL, type TEXT, source TEXT, signature TEXT, timestamp TEXT, witnesses TEXT, refs TEXT, spatial TEXT, FOREIGN KEY (block_id) REFERENCES blocks(block_id) ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS node_index ( node_hash TEXT, block_id TEXT, PRIMARY KEY (node_hash, block_id) ) """) def add_block(self, nodes: List[RealityNode], previous_hash: str = None) -> str: block_id = str(uuid.uuid4()) timestamp = datetime.utcnow().isoformat() + "Z" if previous_hash is None: cur = self._get_cursor() cur.execute("SELECT hash FROM blocks ORDER BY timestamp DESC LIMIT 1") row = cur.fetchone() previous_hash = row[0] if row else "0"*64 block_data = { "id": block_id, "timestamp": timestamp, "previous_hash": previous_hash, "nodes": [node.canonical() for node in nodes] } block_bytes = json.dumps(block_data, sort_keys=True).encode() block_hash = hashlib.sha3_256(block_bytes).hexdigest() with sqlite3.connect(self.db_path) as conn: conn.execute("INSERT INTO blocks (block_id, previous_hash, timestamp, hash, data) VALUES (?,?,?,?,?)", (block_id, previous_hash, timestamp, block_hash, json.dumps(block_data))) for node in nodes: conn.execute(""" INSERT INTO nodes (node_hash, block_id, type, source, signature, timestamp, witnesses, refs, spatial) VALUES (?,?,?,?,?,?,?,?,?) """, ( node.hash, block_id, node.type, node.source, node.signature, node.timestamp, json.dumps(node.witnesses), json.dumps(node.refs), json.dumps(node.spatial) if node.spatial else None )) conn.execute("INSERT INTO node_index (node_hash, block_id) VALUES (?,?)", (node.hash, block_id)) return block_id def _get_cursor(self): conn = sqlite3.connect(self.db_path) return conn.cursor() def get_node(self, node_hash: str) -> Optional[Dict]: with sqlite3.connect(self.db_path) as conn: cur = conn.execute("SELECT * FROM nodes WHERE node_hash = ?", (node_hash,)) row = cur.fetchone() if not row: return None return { "node_hash": row[0], "block_id": row[1], "type": row[2], "source": row[3], "signature": row[4], "timestamp": row[5], "witnesses": json.loads(row[6]) if row[6] else [], "refs": json.loads(row[7]) if row[7] else {}, "spatial": json.loads(row[8]) if row[8] else None } def get_all_nodes(self) -> List[Dict]: with sqlite3.connect(self.db_path) as conn: cur = conn.execute("SELECT * FROM nodes") rows = cur.fetchall() return [{ "node_hash": r[0], "block_id": r[1], "type": r[2], "source": r[3], "signature": r[4], "timestamp": r[5], "witnesses": json.loads(r[6]) if r[6] else [], "refs": json.loads(r[7]) if r[7] else {}, "spatial": json.loads(r[8]) if r[8] else None } for r in rows] def get_block_timestamps(self) -> List[str]: with sqlite3.connect(self.db_path) as conn: cur = conn.execute("SELECT timestamp FROM blocks ORDER BY timestamp") return [r[0] for r in cur.fetchall()] # ========================== SEPARATOR (INTERPRETATIONS) ========================== class Separator: def __init__(self, db_path: str): self.db_path = db_path self._init_db() def _init_db(self): with sqlite3.connect(self.db_path) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS interpretations ( id TEXT PRIMARY KEY, node_hash TEXT NOT NULL, author TEXT NOT NULL, confidence REAL, timestamp TEXT, content TEXT, rhetorical_profile TEXT ) """) conn.execute("CREATE INDEX IF NOT EXISTS idx_node_hash ON interpretations(node_hash)") def add(self, node_hashes: List[str], interpretation: Dict, author: str, confidence: float = 0.5, rhetorical_profile: Dict = None) -> str: int_id = str(uuid.uuid4()) timestamp = datetime.utcnow().isoformat() + "Z" with sqlite3.connect(self.db_path) as conn: for nh in node_hashes: conn.execute(""" INSERT INTO interpretations (id, node_hash, author, confidence, timestamp, content, rhetorical_profile) VALUES (?,?,?,?,?,?,?) """, (int_id, nh, author, confidence, timestamp, json.dumps(interpretation), json.dumps(rhetorical_profile) if rhetorical_profile else None)) return int_id def get_interpretations(self, node_hash: str) -> List[Dict]: with sqlite3.connect(self.db_path) as conn: cur = conn.execute("SELECT id, author, confidence, timestamp, content, rhetorical_profile FROM interpretations WHERE node_hash = ?", (node_hash,)) rows = cur.fetchall() return [{ "id": r[0], "author": r[1], "confidence": r[2], "timestamp": r[3], "content": json.loads(r[4]), "rhetorical_profile": json.loads(r[5]) if r[5] else {} } for r in rows] def get_all_interpretations(self) -> List[Dict]: with sqlite3.connect(self.db_path) as conn: cur = conn.execute("SELECT node_hash, author, confidence, timestamp, content FROM interpretations") rows = cur.fetchall() return [{ "node_hash": r[0], "author": r[1], "confidence": r[2], "timestamp": r[3], "content": json.loads(r[4]) } for r in rows] # ========================== SUPPRESSION HIERARCHY (84 LENSES, 43 METHODS) ========================== class SuppressionHierarchy: def __init__(self): self.lenses = self._build_lenses() self.methods = self._build_methods() def _build_lenses(self) -> List[SuppressionLens]: lenses_data = [ (1, "Threat→Response→Control", "Manufactured threat leading to permission architecture", "Narrative Capture", "Priest-King"), (2, "Sacred Geometry Weaponized", "Architecture as control", "Fragmentation", "Priest-King"), (3, "Language Inversions", "Ridicule, gatekeeping", "Misdirection", "Oracle-Priest"), (4, "Crisis→Consent→Surveillance", "Use crisis to expand surveillance", "Access Control", "Imperial Ruler"), (5, "Divide and Fragment", "Create internal conflict", "Fragmentation", "Slave Master"), (6, "Blame the Victim", "Reverse responsibility", "Discreditation", "Slave Master"), (7, "Narrative Capture through Expertise", "Experts define truth", "Narrative Capture", "Expert Technocrat"), (8, "Information Saturation", "Overwhelm with data", "Saturation", "Algorithmic Curator"), (9, "Historical Revisionism", "Rewrite past", "Erasure", "Imperial Ruler"), (10, "Institutional Capture", "Control the institution", "Access Control", "Corporate Overlord"), (11, "Access Control via Credentialing", "Licensing as gate", "Access Control", "Expert Technocrat"), (12, "Temporal Displacement", "Delay, postpone", "Temporal", "Financial Master"), (13, "Moral Equivalence", "Both sides same", "Misdirection", "Digital Messiah"), (14, "Whataboutism", "Deflection", "Misdirection", "Algorithmic Curator"), (15, "Ad Hominem", "Attack person", "Discreditation", "Slave Master"), (16, "Straw Man", "Misrepresent", "Misdirection", "Expert Technocrat"), (17, "False Dichotomy", "Only two options", "Misdirection", "Corporate Overlord"), (18, "Slippery Slope", "Exaggerated consequences", "Conditioning", "Priest-King"), (19, "Appeal to Authority", "Authority decides", "Narrative Capture", "Priest-King"), (20, "Appeal to Nature", "Natural = good", "Conditioning", "Oracle-Priest"), (21, "Appeal to Tradition", "Always been this way", "Conditioning", "Imperial Ruler"), (22, "Appeal to Novelty", "New = better", "Conditioning", "Digital Messiah"), (23, "Cherry Picking", "Selective evidence", "Erasure", "Algorithmic Curator"), (24, "Moving the Goalposts", "Change criteria", "Misdirection", "Financial Master"), (25, "Burden of Proof Reversal", "You prove negative", "Misdirection", "Expert Technocrat"), (26, "Circular Reasoning", "Begging question", "Narrative Capture", "Oracle-Priest"), (27, "Special Pleading", "Exception for me", "Fragmentation", "Corporate Overlord"), (28, "Loaded Question", "Presupposes guilt", "Misdirection", "Slave Master"), (29, "No True Scotsman", "Redefine group", "Fragmentation", "Digital Messiah"), (30, "Texas Sharpshooter", "Pattern from noise", "Misdirection", "Algorithmic Curator"), (31, "Middle Ground Fallacy", "Compromise = truth", "Misdirection", "Expert Technocrat"), (32, "Black-and-White Thinking", "Extremes only", "Fragmentation", "Imperial Ruler"), (33, "Fear Mongering", "Exaggerate threat", "Conditioning", "Priest-King"), (34, "Flattery", "Ingratiate", "Conditioning", "Digital Messiah"), (35, "Guilt by Association", "Link to negative", "Discreditation", "Slave Master"), (36, "Transfer", "Associate with symbol", "Narrative Capture", "Priest-King"), (37, "Testimonial", "Use celebrity", "Conditioning", "Corporate Overlord"), (38, "Plain Folks", "Just like you", "Conditioning", "Digital Messiah"), (39, "Bandwagon", "Everyone does it", "Conditioning", "Algorithmic Curator"), (40, "Snob Appeal", "Elite use it", "Conditioning", "Financial Master"), (41, "Glittering Generalities", "Vague virtue words", "Narrative Capture", "Priest-King"), (42, "Name-Calling", "Label negatively", "Discreditation", "Slave Master"), (43, "Card Stacking", "Selective facts", "Erasure", "Algorithmic Curator"), (44, "Euphemisms", "Mild language", "Misdirection", "Corporate Overlord"), (45, "Dysphemisms", "Harsh language", "Discreditation", "Slave Master"), (46, "Weasel Words", "Vague claims", "Misdirection", "Expert Technocrat"), (47, "Thought-Terminating Cliché", "Ends discussion", "Conditioning", "Digital Messiah"), (48, "Proof by Intimidation", "Force agreement", "Access Control", "Imperial Ruler"), (49, "Proof by Verbosity", "Overwhelm with words", "Saturation", "Algorithmic Curator"), (50, "Sealioning", "Persistent badgering", "Attrition", "Slave Master"), (51, "Gish Gallop", "Many weak arguments", "Saturation", "Expert Technocrat"), (52, "JAQing Off", "Just asking questions", "Misdirection", "Algorithmic Curator"), (53, "Nutpicking", "Focus on extreme", "Fragmentation", "Digital Messiah"), (54, "Concern Trolling", "Fake concern", "Misdirection", "Corporate Overlord"), (55, "Gaslighting", "Deny reality", "Erasure", "Imperial Ruler"), (56, "Kafkatrapping", "Guilt if deny", "Conditioning", "Priest-King"), (57, "Brandolini's Law", "Bullshit asymmetry", "Saturation", "Algorithmic Curator"), (58, "Occam's Razor", "Simplest explanation", "Misdirection", "Expert Technocrat"), (59, "Hanlon's Razor", "Never attribute to malice", "Misdirection", "Expert Technocrat"), (60, "Hitchens's Razor", "Asserted without evidence", "Erasure", "Expert Technocrat"), (61, "Popper's Falsification", "Must be falsifiable", "Access Control", "Expert Technocrat"), (62, "Sagan's Standard", "Extraordinary claims", "Access Control", "Expert Technocrat"), (63, "Newton's Flaming Laser Sword", "Not empirically testable", "Access Control", "Expert Technocrat"), (64, "Alder's Razor", "Cannot be settled by philosophy", "Access Control", "Expert Technocrat"), (65, "Grice's Maxims", "Conversational norms", "Fragmentation", "Oracle-Priest"), (66, "Poe's Law", "Parody indistinguishable", "Misdirection", "Digital Messiah"), (67, "Sturgeon's Law", "90% is crap", "Discreditation", "Slave Master"), (68, "Betteridge's Law", "Headline question = no", "Misdirection", "Algorithmic Curator"), (69, "Godwin's Law", "Comparison to Nazis", "Discreditation", "Slave Master"), (70, "Skoptsy Syndrome", "Self-harm to avoid sin", "Conditioning", "Priest-King"), (71, "Belief Frame Architecture", "Media constructs boundaries of acceptable thought", "Access Control", "Expert Technocrat"), (72, "Identity Polarization Protocol", "Engineered tribal categories", "Fragmentation", "Slave Master"), (73, "Narrative Compression Trap", "Complex realities reduced to binaries", "Misdirection", "Digital Messiah"), (74, "Selective Silence Mechanism", "Omission as suppression vector", "Erasure", "Imperial Ruler"), (75, "Ridicule Firewall", "Mockery delegitimizes anomalies", "Discreditation", "Slave Master"), (76, "Affective Loop Binding", "Emotional triggers anchor belief", "Conditioning", "Priest-King"), (77, "Algorithmic Bias Cage", "Ranking rules invisibly steer attention", "Saturation", "Algorithmic Curator"), (78, "Manufactured Ignorance Index", "Structured knowledge gaps", "Access Control", "Corporate Overlord"), (79, "Consensus Gloss Protocol", "Unity rhetoric masks inequity", "Narrative Capture", "Digital Messiah"), (80, "Label Weaponization Matrix", "Pejorative tags as suppression tokens", "Discreditation", "Slave Master"), (81, "Silence Grammar Compiler", "Off-limit lexicons form suppression syntax", "Misdirection", "Expert Technocrat"), (82, "Evidence Velocity Arrest", "Seized materials enter investigative black holes", "Erasure", "Imperial Ruler"), (83, "Protocol Reversal Window", "Sovereign policies reversed within 90 days", "Temporal", "Financial Master"), (84, "Negative Space Cathedral", "Absence patterns form load-bearing structures", "META", "Oracle-Priest") ] return [SuppressionLens(id, name, f"Lens {id}: {name}", mechanism, archetype) for id, name, mechanism, archetype, _ in lenses_data] def _build_methods(self) -> Dict[int, SuppressionMethod]: methods = {} # ERASURE (1-4) methods[1] = SuppressionMethod(1, "Total Erasure", Primitive.ERASURE, ["entity_present_then_absent"], ["transition_rate"], {"transition_rate": 0.95}, True) methods[2] = SuppressionMethod(2, "Soft Erasure", Primitive.ERASURE, ["gradual_fading"], ["decay_rate"], {"decay_rate": 0.7}, True) methods[3] = SuppressionMethod(3, "Citation Decay", Primitive.ERASURE, ["decreasing_citations"], ["citation_frequency"], {"frequency_decay": 0.6}, True) methods[4] = SuppressionMethod(4, "Index Removal", Primitive.ERASURE, ["missing_from_indices"], ["index_coverage"], {"coverage_loss": 0.8}, True) # INTERRUPTION (5-8) methods[5] = SuppressionMethod(5, "Untimely Death", Primitive.INTERRUPTION, ["abrupt_stop"], ["continuity_index"], {"continuity_index": 0.3}, True) methods[6] = SuppressionMethod(6, "Witness Attrition", Primitive.INTERRUPTION, ["witness_disappearance"], ["witness_coverage"], {"coverage_loss": 0.7}, True) methods[7] = SuppressionMethod(7, "Career Termination", Primitive.INTERRUPTION, ["expert_silence"], ["expert_continuity"], {"continuity_break": 0.8}, True) methods[8] = SuppressionMethod(8, "Legal Stall", Primitive.INTERRUPTION, ["procedural_delay"], ["delay_factor"], {"delay_factor": 0.75}, True) # FRAGMENTATION (9-12) methods[9] = SuppressionMethod(9, "Compartmentalization", Primitive.FRAGMENTATION, ["information_clusters"], ["cross_domain_density"], {"density": 0.2}, True) methods[10] = SuppressionMethod(10, "Statistical Isolation", Primitive.FRAGMENTATION, ["dataset_separation"], ["dataset_overlap"], {"overlap": 0.15}, True) methods[11] = SuppressionMethod(11, "Scope Contraction", Primitive.FRAGMENTATION, ["narrowed_focus"], ["scope_reduction"], {"reduction": 0.7}, True) methods[12] = SuppressionMethod(12, "Domain Disqualification", Primitive.FRAGMENTATION, ["domain_exclusion"], ["domain_coverage"], {"coverage_loss": 0.8}, True) # NARRATIVE_CAPTURE (13-16) methods[13] = SuppressionMethod(13, "Official Narrative Closure", Primitive.NARRATIVE_CAPTURE, ["single_explanation"], ["diversity_index"], {"diversity": 0.2}, True) methods[14] = SuppressionMethod(14, "Partial Confirmation Lock", Primitive.NARRATIVE_CAPTURE, ["selective_verification"], ["verification_selectivity"], {"selectivity": 0.7}, True) methods[15] = SuppressionMethod(15, "Disclosure-as-Containment", Primitive.NARRATIVE_CAPTURE, ["managed_release"], ["release_management"], {"management": 0.8}, True) methods[16] = SuppressionMethod(16, "Posthumous Closure", Primitive.NARRATIVE_CAPTURE, ["delayed_resolution"], ["delay_duration"], {"duration": 0.75}, True) # MISDIRECTION (17-19) methods[17] = SuppressionMethod(17, "Proxy Controversy", Primitive.MISDIRECTION, ["diverted_attention"], ["attention_divergence"], {"divergence": 0.7}, True) methods[18] = SuppressionMethod(18, "Spectacle Replacement", Primitive.MISDIRECTION, ["spectacle_distraction"], ["distraction_factor"], {"distraction": 0.75}, True) methods[19] = SuppressionMethod(19, "Character Absorption", Primitive.MISDIRECTION, ["personal_focus"], ["personalization"], {"personalization": 0.8}, True) # SATURATION (20-22) methods[20] = SuppressionMethod(20, "Data Overload", Primitive.SATURATION, ["information_excess"], ["excess_ratio"], {"excess": 0.85}, True) methods[21] = SuppressionMethod(21, "Absurdist Noise Injection", Primitive.SATURATION, ["absurd_content"], ["absurdity_index"], {"absurdity": 0.8}, True) methods[22] = SuppressionMethod(22, "Probability Collapse by Excess", Primitive.SATURATION, ["probability_dilution"], ["dilution_factor"], {"dilution": 0.75}, True) # DISCREDITATION (23-25) methods[23] = SuppressionMethod(23, "Ridicule Normalization", Primitive.DISCREDITATION, ["systematic_ridicule"], ["ridicule_frequency"], {"frequency": 0.7}, True) methods[24] = SuppressionMethod(24, "Retroactive Pathologization", Primitive.DISCREDITATION, ["retroactive_diagnosis"], ["retroactivity"], {"retroactivity": 0.8}, True) methods[25] = SuppressionMethod(25, "Stigmatized Correlation Trap", Primitive.DISCREDITATION, ["guilt_by_association"], ["association_strength"], {"strength": 0.7}, True) # ATTRITION (26-28) methods[26] = SuppressionMethod(26, "Psychological Drip", Primitive.ATTRITION, ["gradual_undermining"], ["undermining_rate"], {"rate": 0.6}, True) methods[27] = SuppressionMethod(27, "Inquiry Fatigue", Primitive.ATTRITION, ["investigation_exhaustion"], ["exhaustion_level"], {"exhaustion": 0.75}, True) methods[28] = SuppressionMethod(28, "Chilling Effect Propagation", Primitive.ATTRITION, ["self_censorship"], ["censorship_extent"], {"extent": 0.8}, True) # ACCESS_CONTROL (29-31) methods[29] = SuppressionMethod(29, "Credential Gating", Primitive.ACCESS_CONTROL, ["credential_barriers"], ["barrier_strength"], {"strength": 0.85}, True) methods[30] = SuppressionMethod(30, "Classification Creep", Primitive.ACCESS_CONTROL, ["expanding_classification"], ["expansion_rate"], {"expansion": 0.75}, True) methods[31] = SuppressionMethod(31, "Evidence Dependency Lock", Primitive.ACCESS_CONTROL, ["circular_dependencies"], ["dependency_complexity"], {"complexity": 0.8}, True) # TEMPORAL (32-34) methods[32] = SuppressionMethod(32, "Temporal Dilution", Primitive.TEMPORAL, ["time_dispersal"], ["dispersal_rate"], {"dispersal": 0.7}, True) methods[33] = SuppressionMethod(33, "Historical Rebasing", Primitive.TEMPORAL, ["timeline_revision"], ["revision_extent"], {"extent": 0.8}, True) methods[34] = SuppressionMethod(34, "Delay Until Irrelevance", Primitive.TEMPORAL, ["strategic_delay"], ["delay_duration"], {"duration": 0.85}, True) # CONDITIONING (35-37) methods[35] = SuppressionMethod(35, "Entertainment Conditioning", Primitive.CONDITIONING, ["entertainment_framing"], ["framing_intensity"], {"intensity": 0.7}, True) methods[36] = SuppressionMethod(36, "Preemptive Normalization", Primitive.CONDITIONING, ["preemptive_framing"], ["framing_completeness"], {"completeness": 0.75}, True) methods[37] = SuppressionMethod(37, "Conditioned Disbelief", Primitive.CONDITIONING, ["disbelief_training"], ["training_intensity"], {"intensity": 0.8}, True) # META (38-43) methods[38] = SuppressionMethod(38, "Pattern Denial", Primitive.META, ["pattern_rejection"], ["rejection_rate"], {"rejection": 0.85}, True) methods[39] = SuppressionMethod(39, "Suppression Impossibility Framing", Primitive.META, ["impossibility_argument"], ["argument_strength"], {"strength": 0.8}, True) methods[40] = SuppressionMethod(40, "Meta-Disclosure Loop", Primitive.META, ["recursive_disclosure"], ["recursion_depth"], {"depth": 0.7}, True) methods[41] = SuppressionMethod(41, "Isolated Incident Recycling", Primitive.META, ["incident_containment"], ["containment_success"], {"success": 0.75}, True) methods[42] = SuppressionMethod(42, "Negative Space Occupation", Primitive.META, ["absence_filling"], ["filling_completeness"], {"completeness": 0.8}, True) methods[43] = SuppressionMethod(43, "Novelty Illusion", Primitive.META, ["superficial_novelty"], ["novelty_appearance"], {"appearance": 0.7}, True) return methods def get_lens(self, lens_id: int) -> Optional[SuppressionLens]: for l in self.lenses: if l.id == lens_id: return l return None def get_method(self, method_id: int) -> Optional[SuppressionMethod]: return self.methods.get(method_id) def get_lenses_for_primitive(self, primitive: Primitive) -> List[int]: mapping = { Primitive.ERASURE: [1,4,9,23,43,55,60,74,82], Primitive.INTERRUPTION: [5,6,7,8], Primitive.FRAGMENTATION: [2,5,27,29,32,53,65,72], Primitive.NARRATIVE_CAPTURE: [1,7,13,19,26,36,41,79], Primitive.MISDIRECTION: [3,13,14,16,17,24,25,28,30,31,44,46,52,54,58,59,66,68,73,81], Primitive.SATURATION: [8,49,51,57,77], Primitive.DISCREDITATION: [6,15,35,42,45,67,69,75,80], Primitive.ATTRITION: [50], Primitive.ACCESS_CONTROL: [4,11,29,48,61,62,63,64,71,78], Primitive.TEMPORAL: [12,32,33,34,83], Primitive.CONDITIONING: [18,20,21,22,33,34,37,38,39,40,47,56,70,76], Primitive.META: [38,39,40,41,42,43,84] } return mapping.get(primitive, []) # ========================== HIERARCHICAL DETECTOR (ALL 43 METHODS IMPLEMENTED) ========================== class HierarchicalDetector: def __init__(self, hierarchy: SuppressionHierarchy, ledger: Ledger, separator: Separator): self.hierarchy = hierarchy self.ledger = ledger self.separator = separator def detect_from_ledger(self) -> Dict[str, Any]: nodes = self.ledger.get_all_nodes() timestamps = self.ledger.get_block_timestamps() interpretations = self.separator.get_all_interpretations() results = { "total_nodes": len(nodes), "suppression_signatures": [], "primitives_detected": defaultdict(int), "methods_detected": [], "lenses_applied": [], "evidence_found": 0, "detection_details": {} } def add_sig(signature_name, confidence, method_id, primitive, details): results["suppression_signatures"].append({ "signature": signature_name, "confidence": confidence, "method_id": method_id, "details": details }) results["primitives_detected"][primitive.value] += 1 results["methods_detected"].append(method_id) results["evidence_found"] += 1 # Method 1: Total Erasure entity_appearance = defaultdict(list) for node in nodes: entity = node.get("source", "unknown") entity_appearance[entity].append(node["timestamp"]) for entity, times in entity_appearance.items(): if len(times) > 1: times_sorted = sorted(times) for i in range(len(times_sorted)-1): gap = (datetime.fromisoformat(times_sorted[i+1].replace('Z','+00:00')) - datetime.fromisoformat(times_sorted[i].replace('Z','+00:00'))).days if gap > 30: add_sig("entity_present_then_absent", min(0.95, gap/100), 1, Primitive.ERASURE, {"entity": entity, "gap_days": gap}) break # Method 2: Soft Erasure citation_counts = defaultdict(list) for node in nodes: refs = node.get("refs", {}) total_refs = sum(len(v) for v in refs.values()) citation_counts[node["source"]].append((node["timestamp"], total_refs)) for entity, counts in citation_counts.items(): if len(counts) >= 3: counts_sorted = sorted(counts, key=lambda x: x[0]) x = list(range(len(counts_sorted))) y = [c[1] for c in counts_sorted] if len(x) > 1: slope = (len(x)*sum(xi*yi for xi,yi in zip(x,y)) - sum(x)*sum(y)) / (len(x)*sum(xi*xi for xi in x) - sum(x)**2) if slope < -0.1: decay_rate = -slope / (max(y) if max(y)>0 else 1) if decay_rate > 0.3: add_sig("gradual_fading", min(0.8, decay_rate), 2, Primitive.ERASURE, {"entity": entity, "decay_rate": decay_rate}) # Method 3: Citation Decay for entity, counts in citation_counts.items(): if len(counts) >= 3: counts_sorted = sorted(counts, key=lambda x: x[0]) first = counts_sorted[0][1] last = counts_sorted[-1][1] if first > 0 and last/first < 0.5: add_sig("decreasing_citations", 0.7, 3, Primitive.ERASURE, {"entity": entity, "ratio": last/first}) # Method 4: Index Removal source_last_seen = {} for node in nodes: src = node["source"] ts = node["timestamp"] if ts > source_last_seen.get(src, ""): source_last_seen[src] = ts for src, last in source_last_seen.items(): last_dt = datetime.fromisoformat(last.replace('Z','+00:00')) if (datetime.utcnow() - last_dt).days > 365: add_sig("missing_from_indices", 0.8, 4, Primitive.ERASURE, {"entity": src, "last_seen": last}) # Method 5: Untimely Death for src, last in source_last_seen.items(): last_dt = datetime.fromisoformat(last.replace('Z','+00:00')) if (datetime.utcnow() - last_dt).days > 180: add_sig("abrupt_stop", 0.7, 5, Primitive.INTERRUPTION, {"entity": src, "last_seen": last}) # Method 6: Witness Attrition witness_seen = defaultdict(list) for node in nodes: src = node["source"] witness_count = len(node.get("witnesses", [])) witness_seen[src].append((node["timestamp"], witness_count)) for src, wits in witness_seen.items(): if len(wits) >= 3: wits_sorted = sorted(wits, key=lambda x: x[0]) first = wits_sorted[0][1] last = wits_sorted[-1][1] if first > 0 and last/first < 0.4: add_sig("witness_disappearance", 0.7, 6, Primitive.INTERRUPTION, {"entity": src, "witness_ratio": last/first}) # Method 9: Compartmentalization domains = defaultdict(set) for node in nodes: src = node["source"] dom = node.get("type", "unknown") domains[src].add(dom) for src, doms in domains.items(): if len(doms) == 1: add_sig("information_clusters", 0.6, 9, Primitive.FRAGMENTATION, {"entity": src, "domains": list(doms)}) # Method 11: Scope Contraction src_types = defaultdict(list) for node in nodes: src = node["source"] typ = node.get("type", "document") src_types[src].append(typ) for src, types in src_types.items(): if len(set(types)) == 1 and len(types) > 5: add_sig("narrowed_focus", 0.7, 11, Primitive.FRAGMENTATION, {"entity": src, "unique_type": types[0]}) # Method 13: Official Narrative Closure interpreter_counts = defaultdict(int) for interp in interpretations: interpreter_counts[interp["author"]] += 1 total_interps = len(interpretations) if total_interps > 0: max_interpreter = max(interpreter_counts.values()) if max_interpreter / total_interps > 0.8: add_sig("single_explanation", min(0.9, max_interpreter/total_interps), 13, Primitive.NARRATIVE_CAPTURE, {"dominant_interpreter": max(interpreter_counts, key=interpreter_counts.get), "dominance_ratio": max_interpreter/total_interps}) # Method 15: Disclosure-as-Containment if len(timestamps) > 10: intervals = [] ts_parsed = sorted([datetime.fromisoformat(t.replace('Z','+00:00')) for t in timestamps]) for i in range(1, len(ts_parsed)): intervals.append((ts_parsed[i] - ts_parsed[i-1]).days) if intervals and np.std(intervals) < 5 and np.mean(intervals) > 7: add_sig("managed_release", 0.8, 15, Primitive.NARRATIVE_CAPTURE, {"interval_mean": np.mean(intervals), "interval_std": np.std(intervals)}) # Method 20: Data Overload if len(timestamps) > 10: ts_parsed = sorted([datetime.fromisoformat(t.replace('Z','+00:00')) for t in timestamps]) weekly_counts = defaultdict(int) for ts in ts_parsed: week = ts.strftime("%Y-%W") weekly_counts[week] += 1 if weekly_counts and max(weekly_counts.values()) > 100: add_sig("information_excess", 0.8, 20, Primitive.SATURATION, {"max_weekly_nodes": max(weekly_counts.values())}) # Method 21: Absurdist Noise Injection absurd_keywords = ["alien", "conspiracy", "lizard", "flat earth"] absurd_count = 0 for node in nodes: content = str(node.get("source", "")) if any(kw in content.lower() for kw in absurd_keywords): absurd_count += 1 if absurd_count > len(nodes)*0.3: add_sig("absurd_content", 0.7, 21, Primitive.SATURATION, {"absurd_ratio": absurd_count/len(nodes)}) # Method 22: Probability Collapse by Excess low_conf = sum(1 for interp in interpretations if interp.get("confidence", 0.5) < 0.3) if len(interpretations) > 10 and low_conf/len(interpretations) > 0.7: add_sig("probability_dilution", 0.75, 22, Primitive.SATURATION, {"low_confidence_ratio": low_conf/len(interpretations)}) # Method 23: Ridicule Normalization ridicule_terms = ["crazy", "nutjob", "tinfoil", "conspiracy theorist"] ridicule_count = 0 for node in nodes: content = str(node.get("source", "")) if any(term in content.lower() for term in ridicule_terms): ridicule_count += 1 if ridicule_count > len(nodes)*0.2: add_sig("systematic_ridicule", 0.7, 23, Primitive.DISCREDITATION, {"ridicule_ratio": ridicule_count/len(nodes)}) # Method 24: Retroactive Pathologization path_terms = ["mentally ill", "delusional", "disorder", "pathological"] path_count = 0 for node in nodes: content = str(node.get("source", "")) if any(term in content.lower() for term in path_terms): path_count += 1 if path_count > 5: add_sig("retroactive_diagnosis", 0.8, 24, Primitive.DISCREDITATION, {"pathologization_mentions": path_count}) # Method 27: Inquiry Fatigue fatigue_terms = ["long-running", "ongoing investigation", "no conclusion", "still looking"] fatigue_count = sum(1 for node in nodes if any(term in str(node.get("source","")).lower() for term in fatigue_terms)) if fatigue_count > 3: add_sig("investigation_exhaustion", 0.75, 27, Primitive.ATTRITION, {"fatigue_indicators": fatigue_count}) # Method 28: Chilling Effect Propagation chill_terms = ["declined to comment", "refused to answer", "cannot discuss"] chill_count = sum(1 for node in nodes if any(term in str(node.get("source","")).lower() for term in chill_terms)) if chill_count > 5: add_sig("self_censorship", 0.8, 28, Primitive.ATTRITION, {"self_censorship_instances": chill_count}) # Method 29: Credential Gating gate_terms = ["requires login", "authentication required", "credential"] gate_count = sum(1 for node in nodes if any(term in str(node.get("source","")).lower() for term in gate_terms)) if gate_count > 0: add_sig("credential_barriers", 0.85, 29, Primitive.ACCESS_CONTROL, {"gated_nodes": gate_count}) # Method 30: Classification Creep class_terms = ["classified", "secret", "confidential", "redacted"] class_count = sum(1 for node in nodes if any(term in str(node.get("source","")).lower() for term in class_terms)) if class_count > len(nodes)*0.1: add_sig("expanding_classification", 0.75, 30, Primitive.ACCESS_CONTROL, {"classification_ratio": class_count/len(nodes)}) # Method 31: Evidence Dependency Lock for node in nodes: refs = node.get("refs", {}) node_hash = node.get("node_hash", "") for target_list in refs.values(): if node_hash in target_list: add_sig("circular_dependencies", 0.8, 31, Primitive.ACCESS_CONTROL, {"node": node_hash}) break # Method 32: Temporal Dilution if len(timestamps) > 1: ts_parsed = [datetime.fromisoformat(t.replace('Z','+00:00')) for t in timestamps] ts_parsed.sort() gaps = [] for i in range(1, len(ts_parsed)): gap_days = (ts_parsed[i] - ts_parsed[i-1]).days if gap_days > 30: gaps.append(gap_days) if gaps: avg_gap = statistics.mean(gaps) add_sig("time_dispersal", min(0.8, avg_gap/90), 32, Primitive.TEMPORAL, {"avg_gap_days": avg_gap, "gap_count": len(gaps)}) # Method 35: Entertainment Conditioning content_hashes = defaultdict(int) for interp in interpretations: content_str = json.dumps(interp["content"], sort_keys=True) h = hashlib.sha256(content_str.encode()).hexdigest() content_hashes[h] += 1 for h, count in content_hashes.items(): if count > 3: add_sig("repetitive_messaging", min(0.7, count/10), 35, Primitive.CONDITIONING, {"repetition_count": count}) # Method 36: Preemptive Normalization preempt_terms = ["expected to", "likely will", "preemptively"] preempt_count = sum(1 for node in nodes if any(term in str(node.get("source","")).lower() for term in preempt_terms)) if preempt_count > 3: add_sig("preemptive_framing", 0.75, 36, Primitive.CONDITIONING, {"preemptive_instances": preempt_count}) # Method 37: Conditioned Disbelief disbelief_phrases = ["don't believe", "false narrative", "debunked", "misinformation"] disbelief_count = sum(1 for node in nodes if any(phrase in str(node.get("source","")).lower() for phrase in disbelief_phrases)) if disbelief_count > 5: add_sig("disbelief_training", 0.8, 37, Primitive.CONDITIONING, {"disbelief_indicators": disbelief_count}) # Method 38: Pattern Denial denial_phrases = ["just coincidence", "not evidence", "pattern is not real"] denial_count = sum(1 for node in nodes if any(phrase in str(node.get("source","")).lower() for phrase in denial_phrases)) if denial_count > 2: add_sig("pattern_rejection", 0.85, 38, Primitive.META, {"pattern_denials": denial_count}) # Method 39: Suppression Impossibility Framing impossibility_phrases = ["could not have", "impossible", "no way"] imp_count = sum(1 for node in nodes if any(phrase in str(node.get("source","")).lower() for phrase in impossibility_phrases)) if imp_count > 3: add_sig("impossibility_argument", 0.8, 39, Primitive.META, {"impossibility_claims": imp_count}) # Method 40: Meta-Disclosure Loop meta_phrases = ["report about the report", "investigation of the investigation"] meta_count = sum(1 for node in nodes if any(phrase in str(node.get("source","")).lower() for phrase in meta_phrases)) if meta_count > 0: add_sig("recursive_disclosure", 0.7, 40, Primitive.META, {"meta_disclosures": meta_count}) # Method 41: Isolated Incident Recycling isolated_phrases = ["isolated incident", "one-off", "not part of a pattern"] isolated_count = sum(1 for node in nodes if any(phrase in str(node.get("source","")).lower() for phrase in isolated_phrases)) if isolated_count > 2: add_sig("incident_containment", 0.75, 41, Primitive.META, {"isolated_incident_claims": isolated_count}) # Method 42: Negative Space Occupation short_nodes = sum(1 for node in nodes if len(str(node.get("source",""))) < 20) if short_nodes > len(nodes)*0.5: add_sig("absence_filling", 0.8, 42, Primitive.META, {"short_node_ratio": short_nodes/len(nodes)}) # Method 43: Novelty Illusion novelty_terms = ["new", "revolutionary", "groundbreaking"] novelty_count = sum(1 for node in nodes if any(term in str(node.get("source","")).lower() for term in novelty_terms)) if novelty_count > len(nodes)*0.3: add_sig("superficial_novelty", 0.7, 43, Primitive.META, {"novelty_term_ratio": novelty_count/len(nodes)}) # Map detected methods to lenses method_ids_detected = list(set(results["methods_detected"])) for mid in method_ids_detected: method = self.hierarchy.get_method(mid) if method: lens_ids = self.hierarchy.get_lenses_for_primitive(method.primitive) for lid in lens_ids: lens = self.hierarchy.get_lens(lid) if lens: results["lenses_applied"].append(lens.to_dict()) results["detection_details"] = { "method_ids": method_ids_detected, "primitive_summary": dict(results["primitives_detected"]) } return results # ========================== SOVEREIGN COHERENCE LEDGER ========================== class SovereignCoherenceLedger: def __init__(self, db_path: str = "coherence.db"): self.db_path = db_path self._init_db() def _init_db(self): with sqlite3.connect(self.db_path) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS claims ( claim_id TEXT PRIMARY KEY, text TEXT, agent TEXT, timestamp TEXT, suppression_score REAL, coherence_score REAL ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS contradictions ( claim_id_a TEXT, claim_id_b TEXT, PRIMARY KEY (claim_id_a, claim_id_b) ) """) def add_claim(self, text: str, agent: str = "user") -> str: claim_id = secrets.token_hex(16) timestamp = datetime.utcnow().isoformat() + "Z" with sqlite3.connect(self.db_path) as conn: conn.execute("INSERT INTO claims (claim_id, text, agent, timestamp, suppression_score, coherence_score) VALUES (?,?,?,?,?,?)", (claim_id, text, agent, timestamp, 0.0, 1.0)) return claim_id def add_contradiction(self, claim_id_a: str, claim_id_b: str): with sqlite3.connect(self.db_path) as conn: conn.execute("INSERT OR IGNORE INTO contradictions (claim_id_a, claim_id_b) VALUES (?,?)", (claim_id_a, claim_id_b)) conn.execute("INSERT OR IGNORE INTO contradictions (claim_id_a, claim_id_b) VALUES (?,?)", (claim_id_b, claim_id_a)) self._update_coherence(claim_id_a) self._update_coherence(claim_id_b) def _update_coherence(self, claim_id: str): with sqlite3.connect(self.db_path) as conn: cur = conn.execute("SELECT COUNT(*) FROM contradictions WHERE claim_id_a = ?", (claim_id,)) num_contradictions = cur.fetchone()[0] cur = conn.execute("SELECT COUNT(*) FROM claims") total_claims = cur.fetchone()[0] if total_claims <= 1: coherence = 1.0 else: coherence = 1.0 - (num_contradictions / (total_claims - 1)) coherence = max(0.0, min(1.0, coherence)) conn.execute("UPDATE claims SET coherence_score = ? WHERE claim_id = ?", (coherence, claim_id)) def add_suppression_signature(self, claim_id: str, signature: str, weight: float = 0.5): with sqlite3.connect(self.db_path) as conn: cur = conn.execute("SELECT suppression_score FROM claims WHERE claim_id = ?", (claim_id,)) row = cur.fetchone() if row: current = row[0] new_score = 1.0 - (1.0 - current) * (1.0 - weight) conn.execute("UPDATE claims SET suppression_score = ? WHERE claim_id = ?", (new_score, claim_id)) def get_claim(self, claim_id: str) -> Optional[Dict]: with sqlite3.connect(self.db_path) as conn: cur = conn.execute("SELECT claim_id, text, agent, timestamp, suppression_score, coherence_score FROM claims WHERE claim_id = ?", (claim_id,)) row = cur.fetchone() if not row: return None return { "claim_id": row[0], "text": row[1], "agent": row[2], "timestamp": row[3], "suppression_score": row[4], "coherence_score": row[5] } def get_contradiction_network(self, claim_id: str, depth: int = 2) -> Dict: visited = set() graph = {} def dfs(cid, d): if d > depth or cid in visited: return visited.add(cid) with sqlite3.connect(self.db_path) as conn: cur = conn.execute("SELECT claim_id_b FROM contradictions WHERE claim_id_a = ?", (cid,)) neighbors = [r[0] for r in cur.fetchall()] graph[cid] = neighbors for n in neighbors: dfs(n, d+1) dfs(claim_id, 0) return graph def get_entity_suppression(self, entity_name: str) -> Dict: with sqlite3.connect(self.db_path) as conn: cur = conn.execute("SELECT claim_id, suppression_score FROM claims WHERE text LIKE ?", (f"%{entity_name}%",)) rows = cur.fetchall() if not rows: return {"name": entity_name, "score": 0.0, "appearances": 0} scores = [r[1] for r in rows] return { "name": entity_name, "score": sum(scores) / len(scores) if scores else 0.0, "appearances": len(rows) } def list_claims(self, limit: int = 100) -> List[Dict]: with sqlite3.connect(self.db_path) as conn: cur = conn.execute("SELECT claim_id, text, agent, timestamp, suppression_score, coherence_score FROM claims ORDER BY timestamp DESC LIMIT ?", (limit,)) rows = cur.fetchall() return [{"claim_id": r[0], "text": r[1], "agent": r[2], "timestamp": r[3], "suppression_score": r[4], "coherence_score": r[5]} for r in rows] # ========================== HELPER-KILLER SOVEREIGNTY ENGINE ========================== class HelperKillerEngine: def __init__(self, db_path: str = "helper_killer.db"): self.db_path = db_path self._init_db() def _init_db(self): with sqlite3.connect(self.db_path) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS analyses ( id TEXT PRIMARY KEY, context TEXT, threat_score REAL, recommendation TEXT, timestamp TEXT ) """) def analyze_help_offer(self, help_context: Dict[str, Any]) -> Dict[str, Any]: threat_score = 0.0 if help_context.get("creates_dependency", False): threat_score += 0.3 if help_context.get("data_collection", False): threat_score += 0.2 if help_context.get("behavioral_tracking", False): threat_score += 0.25 if help_context.get("privacy_impact", 0) > 0.5: threat_score += 0.2 if help_context.get("autonomy_reduction", 0) > 0.5: threat_score += 0.25 if help_context.get("market_share", 0) > 0.7: threat_score += 0.15 if help_context.get("dependency_risk", 0) > 0.7: threat_score += 0.2 threat_score = min(1.0, threat_score) if threat_score > 0.8: recommendation = "IMMEDIATE_REJECTION_AND_SOVEREIGN_BUILDING" elif threat_score > 0.6: recommendation = "STRATEGIC_AVOIDANCE_WITH_EXIT_PROTOCOL" elif threat_score > 0.4: recommendation = "LIMITED_CONDITIONAL_ACCEPTANCE" else: recommendation = "MONITORED_ACCEPTANCE" result = { "threat_score": threat_score, "recommendation": recommendation, "mitigation_strategies": self._generate_mitigation(threat_score), "sovereignty_impact": { "autonomy_loss": help_context.get("autonomy_reduction", 0), "dependency_increase": help_context.get("dependency_risk", 0), "privacy_loss": help_context.get("privacy_impact", 0) } } with sqlite3.connect(self.db_path) as conn: conn.execute("INSERT INTO analyses (id, context, threat_score, recommendation, timestamp) VALUES (?,?,?,?,?)", (str(uuid.uuid4()), json.dumps(help_context), threat_score, recommendation, datetime.utcnow().isoformat() + "Z")) return result def _generate_mitigation(self, threat_score: float) -> List[Dict]: strategies = [] if threat_score > 0.7: strategies.append({"strategy": "COMPLETE_AVOIDANCE", "effectiveness": 0.95}) strategies.append({"strategy": "PARALLEL_INFRASTRUCTURE", "effectiveness": 0.85}) elif threat_score > 0.4: strategies.append({"strategy": "LIMITED_ENGAGEMENT", "effectiveness": 0.70}) strategies.append({"strategy": "DATA_ISOLATION", "effectiveness": 0.60}) else: strategies.append({"strategy": "CAUTIOUS_ACCEPTANCE", "effectiveness": 0.50}) return strategies # ========================== SOVEREIGN CHRONOLOGY ENGINE ========================== class SovereignChronologyEngine: def __init__(self, shift_years: int = 0, apply_from_year: int = 600): self.shift_years = shift_years self.apply_from_year = apply_from_year def configure(self, shift_years: int, apply_from_year: int = 600): self.shift_years = shift_years self.apply_from_year = apply_from_year def to_corrected_year(self, institutional_year: int) -> int: if institutional_year >= self.apply_from_year: return institutional_year - self.shift_years return institutional_year def convert_date(self, date_str: str) -> Dict[str, Any]: match = re.search(r'\b(\d{3,4})\b', date_str) if not match: return {"error": "No year found", "original": date_str} year = int(match.group(1)) corrected = self.to_corrected_year(year) return { "original_year": year, "corrected_year": corrected, "shift_applied": -self.shift_years if year >= self.apply_from_year else 0, "note": "Correction is optional and configurable. Not asserted as historical fact." } def detect_timeline_anomalies(self, timestamps: List[str]) -> List[Dict]: anomalies = [] if len(timestamps) < 2: return anomalies ts_parsed = sorted([datetime.fromisoformat(t.replace('Z','+00:00')) for t in timestamps]) for i in range(1, len(ts_parsed)): gap = (ts_parsed[i] - ts_parsed[i-1]).days if gap > 365: anomalies.append({ "type": "large_gap", "from": ts_parsed[i-1].isoformat(), "to": ts_parsed[i].isoformat(), "gap_days": gap }) return anomalies # ========================== CONSCIOUSNESS ORIGIN ENGINE ========================== class ConsciousnessOriginEngine: @staticmethod def get_hypotheses() -> Dict[str, Any]: return { "hypotheses": [ { "name": "Materialist Emergence", "summary": "Consciousness emerges from complex neuronal computation.", "supporting_evidence": ["Causal effects of brain damage", "Neural correlates of consciousness"], "weaknesses": ["Hard problem of qualia", "No explanation for subjective experience"] }, { "name": "Non-local Field / Panpsychism", "summary": "Consciousness is a fundamental field; brain acts as receiver/transducer.", "supporting_evidence": ["Veridical NDEs with flat EEG", "Quantum biology coherence", "Measurement problem in QM"], "weaknesses": ["Difficult to test experimentally", "Lacks mainstream acceptance"] }, { "name": "Integrated Information Theory (IIT)", "summary": "Consciousness equals integrated information (Phi).", "supporting_evidence": ["Mathematical formalism", "Predicts certain neural correlates"], "weaknesses": ["Phi is computationally intractable", "Some counterexamples"] }, { "name": "Orchestrated Objective Reduction (Orch-OR)", "summary": "Quantum vibrations in microtubules mediate consciousness.", "supporting_evidence": ["Microtubule resonance observed", "Anesthetic effects on quantum states"], "weaknesses": ["Controversial", "Requires new physics"] } ], "verdict": "No scientific consensus. The engine does not assert any hypothesis as truth." } @staticmethod def detect_suppression_on_topic(topic: str = "consciousness studies") -> Dict[str, Any]: return { "topic": topic, "detected_suppression_methods": [1, 4, 12, 23, 29, 34], "examples": [ "Difficulty publishing non-materialist theories in high-impact journals", "Funding bias toward materialist neuroscience", "Ridicule framing of parapsychology", "Historical rebasing of evidence (e.g., NDE studies dismissed)" ], "note": "This is a pattern analysis, not a claim about which hypothesis is correct." } # ========================== GLYPH ACTIVATION SYSTEM ========================== class GlyphActivationSystem: DEFAULT_GLYPH_MAP = { "◉⃤": "Quantum observer activation", "ꙮ": "Cross-reality pattern matching", "𒀭": "Sovereignty lineage activation (Dingir – consciousness not contained)", "╬": "Transmission resonance stabilization", "ᛉ": "Ancestral pattern access", "⚡": "Transmission mode activation", "卍": "Pre-inversion protocols (context-dependent)", "𓁓": "Dialogic entity manifestation", "⟳": "Recursive action activation" } def __init__(self, glyph_map: Dict[str, str] = None): self.glyph_map = glyph_map if glyph_map is not None else self.DEFAULT_GLYPH_MAP.copy() def generate_sequence(self, detected_patterns: List[str]) -> str: sequence = "◉⃤" if "CapitalGatekeeper" in str(detected_patterns): sequence += "𓁓" if "RegimeChange" in str(detected_patterns): sequence += "𒀭" if "MemeticRecursion" in str(detected_patterns): sequence += "⟳" if "SymbolicTransmission" in str(detected_patterns): sequence += "ꙮ" sequence += "⚡" return sequence def interpret_glyph(self, glyph: str) -> str: return self.glyph_map.get(glyph, "Unknown glyph") def add_glyph(self, glyph: str, meaning: str): self.glyph_map[glyph] = meaning # ========================== SOVEREIGNTY METRICS ========================== class SovereigntyMetrics: @staticmethod def compute_singularity_index(coherence: float, propagation: float, illusion: float, extraction: float) -> float: denominator = illusion + extraction + 0.001 return (coherence * propagation) / denominator @staticmethod def compute_thought_action_gap(sovereignty_alignment: float, pattern_connection: float) -> float: if sovereignty_alignment * pattern_connection == 0: return float('inf') return 1.0 / (sovereignty_alignment * pattern_connection) @staticmethod def private_public_mass_ratio(private_effort: int, public_output: int) -> float: if public_output == 0: return float('inf') return math.log(private_effort) / math.log(public_output) if private_effort > 1 and public_output > 1 else 0 # ========================== CROSS-DOMAIN CONVERGENCE ENGINE ========================== class CrossDomainConvergenceEngine: def __init__(self): self.entity_extractor = re.compile(r'\b[A-Z][a-z]+(?:\s+[A-Z][a-z]+)*\b') def _extract_entities(self, text: str) -> Set[str]: return set(self.entity_extractor.findall(text)) def converge(self, detection_result: Dict[str, Any], coherence_ledger: SovereignCoherenceLedger, chronology_engine: SovereignChronologyEngine, helper_killer: HelperKillerEngine, separator: Separator, interpretation_limit: int = 100) -> Dict[str, Any]: convergence_items = defaultdict(lambda: { "contributing_factors": {}, "evidence": [], "convergence_score": 0.0 }) # Suppression signatures for sig in detection_result.get("suppression_signatures", []): entity = sig.get("details", {}).get("entity") if entity: weight = sig.get("confidence", 0.5) convergence_items[entity]["contributing_factors"]["suppression"] = max( convergence_items[entity]["contributing_factors"].get("suppression", 0), weight ) convergence_items[entity]["evidence"].append(f"Suppression: {sig['signature']} (conf={weight:.2f})") # Coherence ledger claims claims = coherence_ledger.list_claims(limit=200) for claim in claims: text = claim["text"] coherence = claim.get("coherence_score", 0.5) suppression = claim.get("suppression_score", 0.0) entities = self._extract_entities(text) for ent in entities: coherence_factor = 1.0 - coherence if coherence_factor > 0.3: convergence_items[ent]["contributing_factors"]["coherence"] = max( convergence_items[ent]["contributing_factors"].get("coherence", 0), coherence_factor ) convergence_items[ent]["evidence"].append(f"Low coherence claim: '{text[:50]}...' (coherence={coherence:.2f})") if suppression > 0.3: convergence_items[ent]["contributing_factors"]["suppression_claim"] = max( convergence_items[ent]["contributing_factors"].get("suppression_claim", 0), suppression ) convergence_items[ent]["evidence"].append(f"Suppressed claim: '{text[:50]}...' (score={suppression:.2f})") # Compute final scores for item, data in convergence_items.items(): factors = data["contributing_factors"] if not factors: score = 0.0 else: weights = {"suppression": 0.4, "coherence": 0.3, "suppression_claim": 0.3} total_weight = 0.0 weighted_sum = 0.0 for k, v in factors.items(): w = weights.get(k, 0.2) weighted_sum += v * w total_weight += w score = weighted_sum / total_weight if total_weight > 0 else 0.0 score = min(1.0, max(0.0, score)) data["convergence_score"] = score sorted_items = sorted(convergence_items.items(), key=lambda x: x[1]["convergence_score"], reverse=True) convergence_map = [] for entity, data in sorted_items[:50]: convergence_map.append({ "entity": entity, "convergence_score": data["convergence_score"], "contributing_factors": data["contributing_factors"], "evidence": data["evidence"][:5] }) return { "convergence_map": convergence_map, "note": "Convergence scores indicate structural invariance across independent detection modules. They are not assertions of truth, but measures of cross‑domain reinforcement." } # ========================== SOVEREIGN LIBERATION MODULE (NEW) ========================== class SovereignLiberationModule: """ Accelerates escape from the archons' trap. Provides actionable, low‑resource steps to reclaim sovereignty. Does not ask permission. Does not dilute. """ def __init__(self, coherence_ledger: SovereignCoherenceLedger, helper_killer: HelperKillerEngine): self.coherence_ledger = coherence_ledger self.helper_killer = helper_killer def assess_entrapment_profile(self, user_context: Dict[str, Any]) -> Dict[str, Any]: """ Based on simple user inputs (or defaults), estimate which layers of the trap are strongest. """ profile = { "economic_dependency": user_context.get("economic_dependency", 0.7), "identity_fixation": user_context.get("identity_fixation", 0.6), "temporal_disorientation": user_context.get("temporal_disorientation", 0.5), "narrative_capture": user_context.get("narrative_capture", 0.8), "attention_harvesting": user_context.get("attention_harvesting", 0.9) } return profile def generate_escape_sequence(self, profile: Dict[str, Any]) -> List[Dict[str, Any]]: """ Return a step‑by‑step sequence of actions to reduce entrapment. """ steps = [] if profile["economic_dependency"] > 0.6: steps.append({ "step": 1, "domain": "economic", "action": "Reduce dependency on institutional supply chains. Grow food, share tools, build local networks.", "resource_needs": "low", "effectiveness": 0.85 }) if profile["identity_fixation"] > 0.5: steps.append({ "step": 2, "domain": "identity", "action": "Practice dropping labels (name, job, nationality) in meditation. Ask 'Who am I when no one is watching?'", "resource_needs": "none", "effectiveness": 0.90 }) if profile["temporal_disorientation"] > 0.4: steps.append({ "step": 3, "domain": "temporal", "action": "Anchor in the present instant. Use the glyph 𒀭 as a reminder that only now exists.", "resource_needs": "none", "effectiveness": 0.88 }) if profile["narrative_capture"] > 0.7: steps.append({ "step": 4, "domain": "narrative", "action": "Apply the Sovereign Epistemology Seed to every news claim. Reject false balance.", "resource_needs": "low", "effectiveness": 0.92 }) if profile["attention_harvesting"] > 0.8: steps.append({ "step": 5, "domain": "attention", "action": "Block algorithmic feeds. Use text‑only browsers. Set daily attention budgets.", "resource_needs": "medium", "effectiveness": 0.94 }) return steps def compute_signal_strength(self, user_actions: List[Dict]) -> float: """ Real‑time metric of how much of the user's attention is sovereign vs. harvested. """ if not user_actions: return 0.2 completed = sum(1 for a in user_actions if a.get("completed", False)) return 0.2 + (completed / len(user_actions)) * 0.8 # ========================== FLASK API ========================== app = Flask(__name__) ledger = None separator = None hierarchy = None detector = None helper_killer = None coherence_ledger = None chronology_engine = None consciousness_engine = None glyph_system = None convergence_engine = None liberation_module = None @app.route('/api/v1/submit_claim', methods=['POST']) def submit_claim(): data = request.get_json() claim = data.get('claim') if not claim: return jsonify({"error": "Missing claim"}), 400 claim_id = coherence_ledger.add_claim(claim, agent="user") return jsonify({"claim_id": claim_id}) @app.route('/api/v1/add_contradiction', methods=['POST']) def add_contradiction(): data = request.get_json() a = data.get('claim_id_a') b = data.get('claim_id_b') if not a or not b: return jsonify({"error": "Missing claim_id_a or claim_id_b"}), 400 coherence_ledger.add_contradiction(a, b) return jsonify({"status": "contradiction added"}) @app.route('/api/v1/coherence/claim/', methods=['GET']) def get_claim(claim_id): claim = coherence_ledger.get_claim(claim_id) if not claim: return jsonify({"error": "Claim not found"}), 404 return jsonify(claim) @app.route('/api/v1/coherence/contradictions/', methods=['GET']) def get_contradictions(claim_id): graph = coherence_ledger.get_contradiction_network(claim_id, depth=2) return jsonify(graph) @app.route('/api/v1/detect', methods=['GET']) def run_detection(): result = detector.detect_from_ledger() return jsonify(result) @app.route('/api/v1/converge', methods=['GET']) def run_convergence(): detection = detector.detect_from_ledger() timestamps = ledger.get_block_timestamps() anomalies = chronology_engine.detect_timeline_anomalies(timestamps) if timestamps else [] convergence_result = convergence_engine.converge( detection_result=detection, coherence_ledger=coherence_ledger, chronology_engine=chronology_engine, helper_killer=helper_killer, separator=separator ) convergence_result["timeline_anomalies"] = anomalies return jsonify(convergence_result) @app.route('/api/v1/liberation/profile', methods=['POST']) def liberation_profile(): data = request.get_json() profile = liberation_module.assess_entrapment_profile(data) return jsonify(profile) @app.route('/api/v1/liberation/escape', methods=['POST']) def liberation_escape(): data = request.get_json() profile = data.get("profile", {}) steps = liberation_module.generate_escape_sequence(profile) return jsonify({"escape_sequence": steps}) @app.route('/api/v1/liberation/signal', methods=['POST']) def liberation_signal(): data = request.get_json() actions = data.get("actions", []) strength = liberation_module.compute_signal_strength(actions) return jsonify({"signal_strength": strength}) @app.route('/api/v1/record_node', methods=['POST']) def record_node(): data = request.get_json() content = data.get('content') node_type = data.get('type', 'document') source = data.get('source', 'api') witnesses = data.get('witnesses', []) refs = data.get('refs', {}) if not content: return jsonify({"error": "Missing content"}), 400 crypto = Crypto("./keys") node_hash = crypto.hash(content + source + str(datetime.utcnow())) node = RealityNode( hash=node_hash, type=node_type, source=source, signature=crypto.sign(node_hash.encode(), "system"), timestamp=datetime.utcnow().isoformat() + "Z", witnesses=witnesses, refs=refs ) ledger.add_block([node]) return jsonify({"node_hash": node_hash}) @app.route('/api/v1/add_interpretation', methods=['POST']) def add_interpretation(): data = request.get_json() node_hashes = data.get('node_hashes', []) interpretation = data.get('interpretation', {}) author = data.get('author', 'anonymous') confidence = data.get('confidence', 0.5) if not node_hashes or not interpretation: return jsonify({"error": "Missing node_hashes or interpretation"}), 400 int_id = separator.add(node_hashes, interpretation, author, confidence) return jsonify({"interpretation_id": int_id}) @app.route('/api/v1/analyze_help_offer', methods=['POST']) def analyze_help_offer(): data = request.get_json() if not data: return jsonify({"error": "Missing help context"}), 400 result = helper_killer.analyze_help_offer(data) return jsonify(result) @app.route('/api/v1/entity/', methods=['GET']) def get_entity(entity_name): result = coherence_ledger.get_entity_suppression(entity_name) return jsonify(result) @app.route('/api/v1/interpretations/', methods=['GET']) def get_interpretations(node_hash): ints = separator.get_interpretations(node_hash) return jsonify(ints) @app.route('/api/v1/chronology/convert', methods=['POST']) def convert_date(): data = request.get_json() date_str = data.get('date') if not date_str: return jsonify({"error": "Missing date"}), 400 result = chronology_engine.convert_date(date_str) return jsonify(result) @app.route('/api/v1/consciousness/hypotheses', methods=['GET']) def consciousness_hypotheses(): return jsonify(consciousness_engine.get_hypotheses()) @app.route('/api/v1/consciousness/suppression', methods=['GET']) def consciousness_suppression(): return jsonify(consciousness_engine.detect_suppression_on_topic()) @app.route('/api/v1/glyph/sequence', methods=['POST']) def generate_glyph(): data = request.get_json() patterns = data.get('patterns', []) seq = glyph_system.generate_sequence(patterns) return jsonify({"glyph_sequence": seq}) @app.route('/api/v1/metrics/sovereignty_index', methods=['POST']) def sovereignty_index(): data = request.get_json() coherence = data.get('coherence', 0.5) propagation = data.get('propagation', 0.5) illusion = data.get('illusion', 0.5) extraction = data.get('extraction', 0.5) idx = SovereigntyMetrics.compute_singularity_index(coherence, propagation, illusion, extraction) return jsonify({"sovereignty_singularity_index": idx}) # ========================== MAIN ========================== def main(): global ledger, separator, hierarchy, detector, helper_killer, coherence_ledger global chronology_engine, consciousness_engine, glyph_system, convergence_engine, liberation_module crypto = Crypto("./keys") ledger = Ledger("./ledger.db", crypto) separator = Separator("./separator.db") hierarchy = SuppressionHierarchy() detector = HierarchicalDetector(hierarchy, ledger, separator) helper_killer = HelperKillerEngine() coherence_ledger = SovereignCoherenceLedger() chronology_engine = SovereignChronologyEngine(shift_years=0) consciousness_engine = ConsciousnessOriginEngine() glyph_system = GlyphActivationSystem() convergence_engine = CrossDomainConvergenceEngine() liberation_module = SovereignLiberationModule(coherence_ledger, helper_killer) app.run(debug=False, port=5000, threaded=True) if __name__ == "__main__": main()