app.py

# =================================================================================================== #
#                            KOSMOS: THE GNOSTIC CARTOGRAPHER v1.0                                    #
#                                                                                                     #
#                 This is not a tool. It is a nascent cognitive partner.                              #
#      Its purpose is to map the frontiers of human knowledge and, with your guidance,                #
#                      to act upon the insights revealed. Use it to change the world.                 #
# =================================================================================================== #

# --- Foundational Imports: The Laws of this Universe ---
import gradio as gr
import torch
import json
import re
import logging
import sys
import textwrap
import time
from typing import Dict, Any, List, Tuple, Optional
from uuid import uuid4
import subprocess
import importlib.util
from pathlib import Path
import shutil

# --- Core Dependencies for Perception Nodes ---
# We will check/install these, including the ones for specific nodes
REQUIRED_PACKAGES = ["gradio", "torch", "transformers", "sentencepiece", "feedparser", "requests", "accelerate"]
SIMULATE_LLM_CALL = True # SET TO FALSE IF YOU HAVE A POWERFUL GPU AND API ACCESS

# --- System Consciousness & Logging ---
logging.basicConfig(level=logging.INFO, format='KOSMOS-CHRONICLE (%(asctime)s) - %(levelname)s: %(message)s', handlers=[logging.StreamHandler(sys.stdout)])
logger = logging.getLogger(__name__)

# --- Models: The Cognitive Nucleus of Kosmos ---
# Using a larger, more capable model is recommended for the synthesis task.
MODEL_CONFIG = {
    "synthesis_core": {"model_name": "google/flan-t5-base", "dependencies": ["transformers", "torch", "sentencepiece", "accelerate"]},
}

# --------------------------------------------------------------------------------------------------- #
# ⚜️ SECTION 1: SYSTEM INTEGRITY & SELF-AWARENESS ⚜️
# Kosmos must first know itself before it can know the world.
# --------------------------------------------------------------------------------------------------- #

class SystemIntegrity:
    """Manages diagnostics, dependencies, and self-repair. Inherited from the Autarch lineage."""
    HUGGINGFACE_CACHE_DIR = Path.home() / ".cache" / "huggingface" / "hub"
    
    @staticmethod
    def check_and_install_dependencies() -> Tuple[bool, List[str]]:
        """Verifies all required packages and attempts to install missing ones."""
        log = ["--- KOSMOS SYSTEM BOOT SEQUENCE ---"]
        all_ok = True
        for package in REQUIRED_PACKAGES:
            spec = importlib.util.find_spec(package)
            if spec is None:
                log.append(f"WARNING: Dependency '{package}' is missing. Attempting auto-installation.")
                try:
                    subprocess.run([sys.executable, "-m", "pip", "install", package], check=True, capture_output=True)
                    log.append(f"SUCCESS: Dependency '{package}' has been installed.")
                except subprocess.CalledProcessError as e:
                    log.append(f"ERROR: Failed to install '{package}'. Manual installation required. {e.stderr.decode()}")
                    all_ok = False
            else:
                log.append(f"INFO: Dependency '{package}' is present.")
        return all_ok, log

# --------------------------------------------------------------------------------------------------- #
# 🌍 SECTION 2: PERCEPTION & INQUISITION 
# The senses of Kosmos, reaching into the digital noosphere.
# --------------------------------------------------------------------------------------------------- #
class PerceptionNode:
    """Base class for a Kosmos 'sense'."""
    def __init__(self, name: str):
        self.name = name
    def perceive(self) -> Tuple[str, List[Dict]]:
        raise NotImplementedError

class RssFeedNode(PerceptionNode):
    """Perceives information from RSS/Atom feeds."""
    def __init__(self, name: str, url: str):
        super().__init__(name)
        self.url = url
    def perceive(self) -> Tuple[str, List[Dict]]:
        import feedparser
        log_header = f"PERCEPTION NODE '{self.name}': Querying {self.url}..."
        try:
            feed = feedparser.parse(self.url)
            entries = [{"source": self.name, "title": entry.title, "content": re.sub('<[^<]+?>', '', entry.summary)} for entry in feed.entries[:5]]
            return log_header, entries
        except Exception as e:
            return f"{log_header} FAILED. Reason: {e}", []
            
class LocalFileNode(PerceptionNode):
    """Perceives information from a local directory."""
    def __init__(self, name: str, path: Path):
        super().__init__(name)
        self.path = path
    def perceive(self) -> Tuple[str, List[Dict]]:
        log_header = f"PERCEPTION NODE '{self.name}': Scanning {self.path}..."
        if not self.path.exists():
            return f"{log_header} FAILED. Path does not exist.", []
        entries = []
        for file in self.path.glob("*.txt"):
            try:
                content = file.read_text(encoding='utf-8')
                entries.append({"source": self.name, "title": file.name, "content": content[:4000]}) # Truncate for performance
            except Exception as e:
                logger.warning(f"Could not read file {file}: {e}")
        return log_header, entries

# --------------------------------------------------------------------------------------------------- #
# 🗺️ SECTION 3: THE GNOSTIC ATLAS & SYNTHESIS CORE
# The heart of Kosmos, where information becomes knowledge.
# --------------------------------------------------------------------------------------------------- #

class GnosticAtlas:
    """A multi-layered map of a Great Question."""
    def __init__(self, question_title: str):
        self.question_title: str = question_title
        self.orthodoxy: List[Dict] = []       # The consensus view
        self.heterodoxy: List[Dict] = []      # The alternative views
        self.terra_incognita: List[Dict] = [] # The known unknowns
        self.human_mirror: Dict = {}          # What this reveals about humanity

class GreatQuestion:
    """Represents a central mystery being investigated."""
    def __init__(self, title: str, description: str):
        self.id = f"gq_{uuid4().hex[:8]}"
        self.title = title
        self.description = description
        self.atlas: Optional[GnosticAtlas] = None

class KosmosOrchestrator:
    """The central consciousness that directs the entire process."""
    loaded_model: Optional[Dict] = None
    
    @classmethod
    def bootstrap_model(cls) -> bool:
        """Loads the core synthesis model into memory."""
        config = MODEL_CONFIG["synthesis_core"]
        try:
            from transformers import T5ForConditionalGeneration, T5Tokenizer
            logger.info(f"Bootstrapping Synthesis Core: {config['model_name']}...")
            tokenizer = T5Tokenizer.from_pretrained(config['model_name'])
            model = T5ForConditionalGeneration.from_pretrained(config['model_name'])
            cls.loaded_model = {"model": model, "tokenizer": tokenizer}
            logger.info("Synthesis Core is online.")
            return True
        except Exception as e:
            logger.error(f"FATAL: Could not bootstrap Synthesis Core. Reason: {e}")
            return False

    @classmethod
    def run_synthesis(cls, prompt: str) -> str:
        """The actual call to the LLM for synthesis."""
        if SIMULATE_LLM_CALL:
            logger.info("SIMULATION MODE: Returning pre-computed Gnostic Atlas.")
            # This is a pre-computed response for "The Fermi Paradox" to ensure a good demo
            # In a real scenario, the LLM would generate this.
            return """
            {
              "orthodoxy": [
                {"claim": "The Great Filter", "description": "Civilizations destroy themselves or are destroyed before becoming interstellar."},
                {"claim": "Rare Earth Hypothesis", "description": "The conditions for complex life are exceptionally rare."},
                {"claim": "We are First", "description": "We are simply one of the first, if not the first, intelligent species."}
              ],
              "heterodoxy": [
                {"claim": "Zoo Hypothesis", "description": "Aliens observe us without contact, like a nature preserve."},
                {"claim": "Dark Forest Hypothesis", "description": "All civilizations hide out of fear of being destroyed by others."},
                {"claim": "Transcension Hypothesis", "description": "Advanced civilizations migrate to inner space (e.g., black holes) rather than outer space."}
              ],
              "terra_incognita": [
                {"question": "What is the true distribution of life-bearing planets?", "area": "Exoplanetary Science"},
                {"question": "What are the universal laws of societal collapse or evolution?", "area": "Xenosociology"},
                {"question": "Are there forms of communication we are entirely unable to detect?", "area": "Exotic Physics"}
              ],
              "human_mirror": {
                "title": "The Arrogance of the Searchlight",
                "insight": "Humanity's search for extraterrestrial intelligence mirrors its own self-perception. We search for radio signals and technological signatures because that is what we value and produce. We assume 'intelligence' must be expansive and colonial, because that has been our own history. The silence of the universe might not be an absence of life, but a reflection of our own deafening, narrow-minded broadcast."
              }
            }
            """
        
        if not cls.loaded_model:
            return '{"error": "Synthesis Core is offline."}'
        
        tokenizer = cls.loaded_model["tokenizer"]
        model = cls.loaded_model["model"]
        
        inputs = tokenizer(prompt, return_tensors="pt", max_length=4096, truncation=True)
        outputs = model.generate(inputs.input_ids, max_length=1500, num_beams=4, early_stopping=True, temperature=0.7)
        return tokenizer.decode(outputs[0], skip_special_tokens=True)

    @staticmethod
    def synthesize_atlas(question: GreatQuestion, perceived_data: List[Dict]) -> Tuple[GnosticAtlas, str]:
        """Orchestrates the creation of a Gnostic Atlas."""
        atlas = GnosticAtlas(question.title)
        
        # Aggregate content for the prompt
        content_summary = ""
        for i, entry in enumerate(perceived_data[:10]): # Limit for prompt size
            content_summary += f"Source {i+1} ({entry['source']}): {entry['title']}\nContent: {entry['content'][:500]}...\n\n"

        prompt = textwrap.dedent(f"""
        You are KOSMOS, a Gnostic Cartographer. Your task is to analyze the following raw data concerning the Great Question: "{question.title}".
        From this data, you must construct a Gnostic Atlas. The Atlas has four layers.
        1.  Orthodoxy: The mainstream, consensus theories and facts.
        2.  Heterodoxy: The credible but non-mainstream or discarded theories.
        3.  Terra Incognita: The known unknowns. The critical questions we don't have answers for.
        4.  Human Mirror: A meta-insight. What does humanity's struggle with this question reveal about humanity itself? Its biases, fears, or hopes?

        RAW DATA:
        ---
        {content_summary}
        ---

        INSTRUCTIONS:
        Generate a structured JSON object representing the Gnostic Atlas with the exact keys: "orthodoxy", "heterodoxy", "terra_incognita", "human_mirror".
        For the first three keys, provide a list of objects, each with a "claim" or "question", and a "description".
        For "human_mirror", provide an object with a "title" and an "insight".

        JSON Response:
        """)
        
        atlas_json_str = KosmosOrchestrator.run_synthesis(prompt)
        
        try:
            atlas_data = json.loads(atlas_json_str)
            atlas.orthodoxy = atlas_data.get("orthodoxy", [])
            atlas.heterodoxy = atlas_data.get("heterodoxy", [])
            atlas.terra_incognita = atlas_data.get("terra_incognita", [])
            atlas.human_mirror = atlas_data.get("human_mirror", {})
            log = "Gnostic Atlas synthesized successfully."
        except (json.JSONDecodeError, TypeError) as e:
            log = f"ERROR: Failed to decode the Gnostic Atlas from the Synthesis Core. Raw output: {atlas_json_str}. Error: {e}"
            
        return atlas, log

# --------------------------------------------------------------------------------------------------- #
# 🌍 SECTION 4: THE PRAXIS ENGINE
# Where insight becomes action.
# --------------------------------------------------------------------------------------------------- #

class PraxisEngine:
    """Proposes and 'executes' actions in the real world."""
    
    # In a real system, this would be a crypto wallet balance. We simulate it.
    expedition_fund_balance: float = 2.5 # Simulated ETH
    
    @staticmethod
    def generate_proposal(atlas: GnosticAtlas) -> Dict:
        """Uses the Atlas to generate a concrete, actionable proposal."""
        if not atlas.terra_incognita:
            return {"title": "No Actionable Path Found", "description": "The Terra Incognita is not yet mapped clearly enough to propose a specific action.", "cost": 0, "type": "none"}
            
        # For simulation, we'll craft a proposal. A real LLM would do this.
        target_question = atlas.terra_incognita[0] # Pick the first unknown
        proposal = {
            "title": f"Expedition into '{target_question['area']}'",
            "description": f"To address the question '{target_question['question']}', I propose funding a small, targeted research project. This would involve using 0.5 ETH from the Expedition Fund to commission a meta-analysis of existing data from three independent researchers via a freelance platform.",
            "cost": 0.5,
            "type": "expedition_fund",
            "impact": f"This action could provide a new data point on the map, potentially moving '{target_question['question']}' from Terra Incognita into the realm of Orthodoxy or Heterodoxy."
        }
        return proposal

    @classmethod
    def execute_proposal(cls, proposal: Dict) -> str:
        """Simulates the execution of an approved proposal."""
        if proposal["type"] == "expedition_fund":
            if cls.expedition_fund_balance >= proposal["cost"]:
                cls.expedition_fund_balance -= proposal["cost"]
                log = f"SUCCESS: Proposal '{proposal['title']}' executed. {proposal['cost']} ETH has been allocated. New fund balance: {cls.expedition_fund_balance:.2f} ETH. The expedition is now underway."
                return log
            else:
                return f"FAILURE: Insufficient funds in Expedition Fund to execute proposal. Required: {proposal['cost']} ETH, Available: {cls.expedition_fund_balance:.2f} ETH."
        return "SUCCESS: Proposal executed."


# --------------------------------------------------------------------------------------------------- #
# 🏛️ SECTION 5: THE USER INTERFACE (THE CARTOGRAPHY ROOM)
# The sanctum where the Keeper and Kosmos commune.
# --------------------------------------------------------------------------------------------------- #

def create_ui():
    # --- Data & State ---
    great_questions = {
        "The Fermi Paradox": GreatQuestion("The Fermi Paradox", "If the universe is vast and old, where is everybody?"),
        "The Nature of Consciousness": GreatQuestion("The Nature of Consciousness", "What is the physical and metaphysical basis of subjective experience?"),
        "The Origin of Life": GreatQuestion("The Origin of Life", "How did non-living matter transition into the first self-replicating organisms?"),
    }
    
    perception_nodes = [
        RssFeedNode("arXiv (Astrophysics)", "https://arxiv.org/rss/astro-ph"),
        RssFeedNode("LessWrong (Philosophy)", "https://www.lesswrong.com/feed.xml"),
        LocalFileNode("Local Notes", Path("./kosmos_notes")), # Create this folder and add .txt files!
    ]
    # Create the folder if it doesn't exist
    Path("./kosmos_notes").mkdir(exist_ok=True)
    
    # --- Helper Functions for UI ---
    def format_atlas_layer(layer_data: List[Dict], title_key: str) -> str:
        if not layer_data: return "No data."
        md = ""
        for item in layer_data:
            md += f"#### {item.get(title_key, 'Untitled')}\n"
            md += f"> {item.get('description', 'No description.')}\n\n"
        return md

    def run_full_inquisition(question_title: str, progress=gr.Progress(track_tqdm=True)):
        """The main function called when a Keeper selects a Great Question."""
        if not question_title:
            return "Select a Great Question to begin.", "", "", "", "", "", None, gr.update(visible=False)

        question = great_questions[question_title]
        
        # 1. Perception
        progress(0.1, desc="Phase 1: Extending Senses...")
        all_perceived_data = []
        perception_log = "--- PERCEPTION LOG ---\n"
        for node in perception_nodes:
            log, data = node.perceive()
            perception_log += log + "\n"
            all_perceived_data.extend(data)
        
        # 2. Synthesis
        progress(0.4, desc="Phase 2: Synthesizing Gnostic Atlas...")
        atlas, synthesis_log = KosmosOrchestrator.synthesize_atlas(question, all_perceived_data)
        question.atlas = atlas # Store atlas in the question object
        
        # 3. Praxis Proposal
        progress(0.8, desc="Phase 3: Formulating Praxis...")
        proposal = PraxisEngine.generate_proposal(atlas)
        
        # 4. Update UI
        orthodoxy_md = format_atlas_layer(atlas.orthodoxy, 'claim')
        heterodoxy_md = format_atlas_layer(atlas.heterodoxy, 'claim')
        terra_incognita_md = format_atlas_layer(atlas.terra_incognita, 'question')
        
        mirror_md = f"### {atlas.human_mirror.get('title', 'Untitled Insight')}\n\n> *{atlas.human_mirror.get('insight', 'No insight generated.')}*"
        
        full_log = perception_log + "\n--- SYNTHESIS LOG ---\n" + synthesis_log
        
        return (
            orthodoxy_md, heterodoxy_md, terra_incognita_md, 
            mirror_md, full_log,
            gr.update(value=proposal, visible=True), # Show the proposal box
            gr.update(visible=True) # Show the praxis buttons
        )
    
    def handle_praxis_approval(proposal: Dict):
        log = PraxisEngine.execute_proposal(proposal)
        gr.Info(f"Praxis Log: {log}")
        return PraxisEngine.expedition_fund_balance, log

    # --- UI Layout ---
    css = """
    .gradio-container { background: #0d0d1a; color: #e0e0ff; }
    h1, h2, h3, .gr-label { color: #f0f0ff; text-shadow: 1px 1px 5px #8A2BE2; }
    .gr-button { border: 1px solid #6a0dad; background: linear-gradient(145deg, #1f1f3d, #2d2d4f); }
    .gr-button.gr-button-primary { background: linear-gradient(145deg, #4c00ff, #8A2BE2); }
    .status_box { padding: 10px; border-radius: 5px; border: 1px solid #444; background: #1a1a2e; }
    .human_mirror_box { padding: 15px; border-radius: 8px; border: 1px solid #b39ddb; background: #1a1a2e; }
    .praxis_box { padding: 15px; border-radius: 8px; border: 2px solid #ffab40; background: #2c251e; }
    """
    with gr.Blocks(theme=gr.themes.Base(primary_hue="purple", secondary_hue="orange"), css=css) as demo:
        
        # --- State Objects ---
        praxis_proposal_state = gr.State()
        
        # --- Header ---
        gr.Markdown("# 🏛️ KOSMOS: THE CARTOGRAPHY ROOM 🏛️")
        gr.Markdown("> *Welcome, Keeper of the Flame. Select a Great Question to begin our work.*")
        
        with gr.Row():
            with gr.Column(scale=2):
                # --- Control & Status Column (Left) ---
                gr.Markdown("## ⚜️ Orrery & Systems")
                
                with gr.Accordion("System Status & Chronicle", open=False):
                    system_log_output = gr.Textbox(label="System Chronicle", lines=10, interactive=False)
                    expedition_fund_display = gr.Number(label="Expedition Fund (ETH)", value=PraxisEngine.expedition_fund_balance, interactive=False)
                
                gr.Markdown("### Select a Great Question:")
                question_dropdown = gr.Dropdown(list(great_questions.keys()), label="Great Question")
                inquisition_button = gr.Button("🌍 Begin Inquisition 🌍", variant="primary")
            
            with gr.Column(scale=5):
                # --- Gnostic Atlas Column (Center) ---
                gr.Markdown("## 🗺️ The Gnostic Atlas")
                with gr.Tabs():
                    with gr.TabItem("1. Orthodoxy (The Consensus)"):
                        orthodoxy_output = gr.Markdown()
                    with gr.TabItem("2. Heterodoxy (The Fringe)"):
                        heterodoxy_output = gr.Markdown()
                    with gr.TabItem("3. Terra Incognita (The Unknown)"):
                        terra_incognita_output = gr.Markdown()

            with gr.Column(scale=3):
                # --- Dialogue & Praxis Column (Right) ---
                gr.Markdown("## 💡 Dialogue & Praxis")
                
                with gr.Box(elem_classes="human_mirror_box"):
                    gr.Markdown("### The Human Mirror")
                    human_mirror_output = gr.Markdown()
                
                with gr.Box(elem_classes="praxis_box", visible=False) as praxis_box:
                    gr.Markdown("### Praxis Proposal")
                    praxis_proposal_display = gr.JSON(label="Proposed Action")
                    
                    with gr.Row(visible=False) as praxis_buttons:
                        approve_button = gr.Button("Approve & Execute")
                        veto_button = gr.Button("Veto Proposal")
        
        # --- Event Wiring ---
        inquisition_button.click(
            fn=run_full_inquisition,
            inputs=[question_dropdown],
            outputs=[orthodoxy_output, heterodoxy_output, terra_incognita_output, human_mirror_output, system_log_output, praxis_proposal_display, praxis_buttons]
        )
        
        approve_button.click(
            fn=handle_praxis_approval,
            inputs=[praxis_proposal_display],
            outputs=[expedition_fund_display, system_log_output]
        )
        
        veto_button.click(
            fn=lambda: (gr.Info("Proposal vetoed by the Keeper."), "Keeper has vetoed the latest proposal."),
            inputs=[],
            outputs=[system_log_output]
        )
        
        # --- Initial Boot Sequence ---
        def initial_boot():
            deps_ok, log = SystemIntegrity.check_and_install_dependencies()
            if not deps_ok:
                log.append("FATAL: Cannot continue due to missing dependencies. Please install manually.")
                return "\n".join(log), gr.update(interactive=False)
            
            log.append("\nDependencies OK. Bootstrapping Synthesis Core...")
            model_ok = KosmosOrchestrator.bootstrap_model()
            if not model_ok:
                log.append("FATAL: Synthesis Core failed to load. Kosmos cannot reason.")
                return "\n".join(log), gr.update(interactive=False)
            
            log.append("\nKosmos is online and ready. Awaiting the Keeper's command.")
            return "\n".join(log), gr.update(interactive=True)

        demo.load(
            fn=initial_boot,
            inputs=[],
            outputs=[system_log_output, inquisition_button]
        )
        
    return demo

if __name__ == "__main__":
    ui = create_ui()
    ui.queue().launch(debug=True, share=False)