""" ๐ŸŒ CASCADE-LATTICE TUI - Textual Terminal User Interface A celebration of provenance. Navigate the lattice like a curious explorer. Run: python -m cascade.tui python cascade/tui.py Features: - Module Explorer: Drill into any module, see actual code - Graph Navigation: Follow relationships, not just hierarchy - Explanation Toggle: "For Dummies" vs "For Scientists" - Live Stats: See your observation counts, genesis root - Interactive Demos: Test functions in real-time "even still, i grow, and yet, I grow still" """ from __future__ import annotations import sys import inspect import importlib from pathlib import Path from datetime import datetime from typing import Dict, List, Any, Optional, Set, Tuple from dataclasses import dataclass, field from textual import on, work from textual.app import App, ComposeResult from textual.binding import Binding from textual.containers import Container, Horizontal, Vertical, ScrollableContainer from textual.screen import Screen from textual.widgets import ( Header, Footer, Static, Button, Label, Tree, TabbedContent, TabPane, RichLog, Markdown, Rule, Switch, Input, ListView, ListItem, Collapsible, ) from textual.widgets.tree import TreeNode from textual.reactive import reactive from textual.message import Message from textual.css.query import NoMatches from rich.text import Text from rich.table import Table from rich.panel import Panel from rich.syntax import Syntax from rich.markdown import Markdown as RichMarkdown # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # MODULE GRAPH - Maps relationships between modules # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• @dataclass class ModuleNode: """A node in the module graph.""" name: str full_path: str doc_dummy: str = "" # Explanation for beginners doc_scientist: str = "" # Technical explanation exports: List[str] = field(default_factory=list) imports_from: List[str] = field(default_factory=list) imported_by: List[str] = field(default_factory=list) children: List[str] = field(default_factory=list) # Sub-modules parent: Optional[str] = None category: str = "core" # core, hold, store, viz, diagnostics, forensics icon: str = "๐Ÿ“ฆ" # The module graph - hand-crafted with love MODULE_GRAPH: Dict[str, ModuleNode] = { # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # ROOT # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• "cascade_lattice": ModuleNode( name="cascade_lattice", full_path="cascade_lattice", icon="๐ŸŒ", category="root", doc_dummy=""" # ๐ŸŒ CASCADE-LATTICE **What is this?** Think of it as a "receipt printer" for AI decisions. Every time an AI makes a choice, cascade-lattice writes it down with a cryptographic signature. Like getting a receipt at a store - you can prove what happened, when, and trace it back to the source. **Why does this matter?** - ๐Ÿ” **Transparency**: See exactly what your AI did - ๐Ÿ›ก๏ธ **Safety**: Humans can pause and review decisions (HOLD) - ๐Ÿ“œ **Provenance**: Trace any output back to its origins - ๐Ÿ”— **Integrity**: Cryptographic proofs prevent tampering **The Genesis Block**: Everything traces back to one root hash. Like the first block in a blockchain, but for AI decisions. """, doc_scientist=""" # CASCADE-LATTICE: Universal AI Provenance Layer A content-addressed, Merkle-authenticated observation framework implementing the HOLD protocol for human-in-the-loop inference control. ## Architecture ``` โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”‚ Genesis Root โ”‚ โ”‚ 89f940c1a4b7 โ”‚ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ฌโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ”‚ โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ผโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ–ผ โ–ผ โ–ผ โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”‚ Receipt โ”‚ โ”‚ Receipt โ”‚ โ”‚ Receipt โ”‚ โ”‚ CID: bafโ”‚ โ”‚ CID: bafโ”‚ โ”‚ CID: bafโ”‚ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ ``` ## Key Concepts - **CID**: Content Identifier (IPFS CIDv1, dag-cbor codec) - **Merkle Root**: SHA-256 hash chain for integrity verification - **HoldPoint**: Freeze-frame of model state for human inspection - **SymbioticAdapter**: Kleene fixed-point signal normalization - **ProvenanceChain**: Linked list of cryptographic attestations ## Module Hierarchy - `core/`: Event, Graph, Provenance, Adapter primitives - `hold/`: HOLD protocol implementation (pause/inspect/override) - `store`: SQLite + IPFS + HuggingFace persistence - `genesis`: Root provenance chain creation - `viz/`: Tape recording and playback - `diagnostics/`: Bug detection, execution monitoring - `forensics/`: Data archaeology, tech fingerprinting """, exports=["Hold", "HoldPoint", "HoldState", "HoldResolution", "Receipt", "Monitor", "SymbioticAdapter", "CausationGraph", "genesis", "observe", "store", "viz", "diagnostics", "forensics"], children=["core", "hold", "store", "genesis", "viz", "diagnostics", "forensics"], ), # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # CORE MODULE # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• "core": ModuleNode( name="core", full_path="cascade_lattice.core", icon="โš™๏ธ", category="core", parent="cascade_lattice", doc_dummy=""" # โš™๏ธ Core Module **What is this?** The engine room of cascade-lattice. Think of it like the foundation of a house - you don't see it, but everything else is built on top of it. **What's inside:** - ๐Ÿ“Š **Event**: A single "thing that happened" - ๐Ÿ”— **Graph**: How events connect to each other - ๐Ÿ” **Provenance**: The chain of custody (who touched what, when) - ๐Ÿค **Adapter**: Translates between different signal formats """, doc_scientist=""" # Core Module (`cascade.core`) Foundational primitives for the observation framework. ## Components ### Event (`core.event`) ```python @dataclass class Event: timestamp: float component: str event_type: str data: Dict[str, Any] event_id: str # Auto-generated UUID ``` ### CausationGraph (`core.graph`) Directed acyclic graph for causal relationship tracking. - `add_event(event)` โ†’ Register event node - `add_link(cause_id, effect_id, strength)` โ†’ Causal edge - `find_path(start, end)` โ†’ Shortest causal path - `get_root_events()` โ†’ Events with no causes ### ProvenanceChain (`core.provenance`) Merkle-authenticated chain of records. - `add_record(input_hash, output_hash, model_hash)` - `verify()` โ†’ Validate chain integrity - `get_lineage()` โ†’ Full audit trail ### SymbioticAdapter (`core.adapter`) Kleene fixed-point signal interpreter. - `interpret(signal)` โ†’ Normalize ANY format to Event - `learned_patterns` โ†’ Discovered signal structures """, exports=["Event", "CausationGraph", "ProvenanceChain", "ProvenanceRecord", "SymbioticAdapter", "compute_merkle_root", "hash_tensor"], children=["event", "graph", "provenance", "adapter"], imported_by=["hold", "store", "genesis", "diagnostics"], ), "event": ModuleNode( name="event", full_path="cascade_lattice.core.event", icon="๐Ÿ“Š", category="core", parent="core", doc_dummy=""" # ๐Ÿ“Š Event **What is this?** A snapshot of "something happened." Like a diary entry: "At 3:42pm, the AI looked at a cat picture and decided it was 90% cat, 10% loaf of bread." **Key parts:** - โฐ **timestamp**: When did it happen? - ๐Ÿท๏ธ **event_type**: What kind of thing? (decision, observation, error) - ๐Ÿ“ **data**: The actual details - ๐Ÿ†” **event_id**: Unique fingerprint for this exact moment """, doc_scientist=""" # Event (`cascade.core.event`) Immutable record of a discrete system occurrence. ```python @dataclass class Event: timestamp: float # Unix epoch (time.time()) component: str # Source subsystem identifier event_type: str # Taxonomy: state_change, decision, observation data: Dict[str, Any] # Arbitrary payload (JSON-serializable) event_id: str # UUID v4, auto-generated source_signal: Any = None # Original input before normalization def to_dict(self) -> Dict[str, Any]: ... @classmethod def from_dict(cls, d: Dict) -> 'Event': ... ``` ## Event Types - `state_change`: Internal state mutation - `decision`: Action selection with confidence - `observation`: External input processing - `hold_point`: HITL pause trigger - `resolution`: Human decision on hold """, exports=["Event"], imports_from=["provenance"], imported_by=["graph", "adapter", "hold.primitives"], ), "graph": ModuleNode( name="graph", full_path="cascade_lattice.core.graph", icon="๐Ÿ•ธ๏ธ", category="core", parent="core", doc_dummy=""" # ๐Ÿ•ธ๏ธ Causation Graph **What is this?** A map of cause and effect. Imagine a detective's board with photos connected by red string. "This event CAUSED that event." You can trace backwards to find the root cause of anything. **What can you do:** - ๐Ÿ” **Trace backwards**: "What caused this bug?" - โžก๏ธ **Trace forwards**: "What did this decision affect?" - ๐ŸŒณ **Find roots**: "Where did it all begin?" - ๐Ÿ“ **Find paths**: "How are these two events connected?" """, doc_scientist=""" # CausationGraph (`cascade.core.graph`) DAG-based causal relationship tracker with temporal ordering. ```python class CausationGraph: def add_event(self, event: Event) -> str: ... def add_link(self, cause_id: str, effect_id: str, strength: float = 1.0) -> None: ... def get_causes(self, event_id: str) -> List[Event]: ... def get_effects(self, event_id: str) -> List[Event]: ... def find_path(self, start_id: str, end_id: str) -> List[str]: ... def get_root_events(self) -> List[Event]: ... def get_terminal_events(self) -> List[Event]: ... def compute_impact(self, event_id: str) -> float: ... ``` ## Link Semantics - `strength โˆˆ [0, 1]`: Causal contribution weight - Multiple causes: `ฮฃ strength_i` need not equal 1 - Temporal constraint: `cause.timestamp < effect.timestamp` ## Traversal - BFS for shortest path - DFS for exhaustive impact analysis - Topological sort for execution order """, exports=["CausationGraph", "CausationChain"], imports_from=["event"], imported_by=["diagnostics", "monitor"], ), "provenance": ModuleNode( name="provenance", full_path="cascade_lattice.core.provenance", icon="๐Ÿ”", category="core", parent="core", doc_dummy=""" # ๐Ÿ” Provenance **What is this?** A chain of custody for AI decisions. Like evidence in a courtroom - you need to prove the evidence wasn't tampered with. Provenance creates a cryptographic trail that proves: "This output really came from this input through this model, and nobody messed with it." **Key functions:** - ๐Ÿงฎ **hash_tensor**: Fingerprint a neural network's weights - ๐Ÿ“ **hash_input**: Fingerprint what went in - ๐Ÿ”— **compute_merkle_root**: Combine all fingerprints into one - โœ… **verify_chain**: Check if anything was tampered with """, doc_scientist=""" # Provenance (`cascade.core.provenance`) Merkle-authenticated provenance tracking with cryptographic verification. ```python def hash_tensor(tensor: np.ndarray) -> str: '''SHA-256 of normalized tensor bytes (first 16 hex chars)''' def hash_params(module: nn.Module) -> str: '''Hash all named_parameters recursively''' def hash_input(data: Any) -> str: '''JSON-serialize then hash''' def compute_merkle_root(hashes: List[str]) -> str: '''Binary tree Merkle root computation''' @dataclass class ProvenanceRecord: input_hash: str output_hash: str model_hash: str timestamp: float parent_merkle: Optional[str] merkle_root: str # Computed from above fields @dataclass class ProvenanceChain: model_id: str session_id: str records: List[ProvenanceRecord] merkle_root: str def add_record(self, input_h, output_h, model_h) -> ProvenanceRecord: ... def verify(self) -> bool: ... def get_lineage(self) -> List[ProvenanceRecord]: ... ``` ## Merkle Tree Structure ``` root / \\ h01 h23 / \\ / \\ h0 h1 h2 h3 ``` """, exports=["ProvenanceChain", "ProvenanceRecord", "ProvenanceTracker", "compute_merkle_root", "hash_tensor", "hash_params", "hash_input", "hash_model", "verify_chain"], imported_by=["genesis", "store", "hold.primitives"], ), "adapter": ModuleNode( name="adapter", full_path="cascade_lattice.core.adapter", icon="๐Ÿค", category="core", parent="core", doc_dummy=""" # ๐Ÿค Symbiotic Adapter **What is this?** A universal translator for AI signals. Different AI models speak different languages - some output dictionaries, some output numpy arrays, some output raw text. The adapter learns to understand ANY format and converts it to a standard Event. **Think of it like:** Google Translate, but for AI outputs. **Magic feature:** It learns patterns as it sees more signals! """, doc_scientist=""" # SymbioticAdapter (`cascade.core.adapter`) Kleene fixed-point signal normalizer implementing universal interpretation. ```python class SymbioticAdapter: def __init__(self): self._patterns: List[SignalPattern] = [] self._signal_count: int = 0 def interpret(self, signal: Any) -> Event: ''' Fixed-point iteration: 1. Try known patterns in order 2. If match, extract structured data 3. If no match, create raw_message wrapper 4. Learn new pattern if structure detected ''' @property def signal_count(self) -> int: ... @property def learned_patterns(self) -> List[str]: ... ``` ## Pattern Learning - Dict with 'embedding' key โ†’ vector extraction - Dict with 'action'/'reward' โ†’ RL transition - numpy.ndarray โ†’ tensor observation - str โ†’ raw message wrapper ## Kleene Semantics Interpretation reaches fixed-point when: - Pattern matching stabilizes - No new patterns discovered in N signals """, exports=["SymbioticAdapter", "SignalPattern"], imports_from=["event"], imported_by=["monitor", "hold.session"], ), # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # HOLD MODULE # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• "hold": ModuleNode( name="hold", full_path="cascade_lattice.hold", icon="โธ๏ธ", category="hold", parent="cascade_lattice", doc_dummy=""" # โธ๏ธ HOLD - Human-in-the-Loop **What is this?** A pause button for AI decisions. Before the AI acts, HOLD freezes everything and asks: "Hey human, here's what I'm about to do. Is this okay?" **You can:** - โœ… **Accept**: "Yes, do it" - โœ๏ธ **Override**: "No, do THIS instead" - โŒ **Cancel**: "Stop everything" - โฐ **Timeout**: If you don't respond, a default happens **Why is this important?** It's the "11th man" - the human who has final say over AI decisions. """, doc_scientist=""" # HOLD Protocol (`cascade.hold`) Human-in-the-loop inference control implementing freeze-frame inspection. ## Architecture ``` Model Forward Pass โ”‚ โ–ผ โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”‚ HoldPoint โ”‚โ—„โ”€โ”€โ”€ Freeze state โ””โ”€โ”€โ”€โ”€โ”€โ”ฌโ”€โ”€โ”€โ”€โ”€โ”˜ โ”‚ โ”Œโ”€โ”€โ”€โ”€โ”ดโ”€โ”€โ”€โ”€โ” โ”‚ PENDING โ”‚ โ””โ”€โ”€โ”€โ”€โ”ฌโ”€โ”€โ”€โ”€โ”˜ โ”‚ โ”Œโ”€โ”€โ”€โ”€โ”€โ”ดโ”€โ”€โ”€โ”€โ”€โ” โ”‚ Human โ”‚ โ”‚ Decision โ”‚ โ””โ”€โ”€โ”€โ”€โ”€โ”ฌโ”€โ”€โ”€โ”€โ”€โ”˜ โ”‚ โ”Œโ”€โ”€โ”€โ”€โ”ดโ”€โ”€โ”€โ”€โ” โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”‚ACCEPTED โ”‚ or โ”‚ OVERRIDDEN โ”‚ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ ``` ## State Machine ``` PENDING โ”€โ”€accept()โ”€โ”€โ–บ ACCEPTED โ”‚ โ”œโ”€โ”€override()โ”€โ”€โ–บ OVERRIDDEN โ”‚ โ”œโ”€โ”€cancel()โ”€โ”€โ–บ CANCELLED โ”‚ โ””โ”€โ”€timeout()โ”€โ”€โ–บ TIMEOUT ``` ## Key Classes - `Hold`: Singleton controller, manages yield_point flow - `HoldPoint`: Immutable snapshot with merkle_root - `HoldResolution`: Result of human decision - `HoldState`: Enum of possible states """, exports=["Hold", "HoldPoint", "HoldResolution", "HoldState", "HoldAwareMixin"], children=["hold.primitives", "hold.session"], imports_from=["core.event", "core.provenance"], imported_by=["store", "viz"], ), "hold.primitives": ModuleNode( name="primitives", full_path="cascade_lattice.hold.primitives", icon="๐Ÿงฑ", category="hold", parent="hold", doc_dummy=""" # ๐Ÿงฑ Hold Primitives **What is this?** The building blocks of HOLD. - **HoldPoint**: A freeze-frame snapshot - What action was the AI going to take? - How confident was it? - What was it looking at? - **HoldState**: Is it waiting? Approved? Overridden? - **HoldResolution**: What did the human decide? """, doc_scientist=""" # Hold Primitives (`cascade.hold.primitives`) Core data structures for the HOLD protocol. ```python class HoldState(Enum): PENDING = 'pending' ACCEPTED = 'accepted' OVERRIDDEN = 'overridden' TIMEOUT = 'timeout' CANCELLED = 'cancelled' @dataclass class HoldPoint: action_probs: np.ndarray # Softmax distribution value: float # V(s) estimate observation: Any # Input state brain_id: str # Source model identifier action_labels: List[str] = None latent: np.ndarray = None # Hidden state attention: np.ndarray = None # Attention weights features: Dict = None # Arbitrary display data imagination: Any = None # World model prediction logits: np.ndarray = None # Pre-softmax scores reasoning: str = None # CoT trace world_prediction: Any = None # Future state prediction # Auto-computed id: str # UUID timestamp: float # Creation time parent_merkle: str = None # Chain linkage merkle_root: str # Integrity hash state: HoldState = PENDING @dataclass class HoldResolution: action: int # Selected action index was_override: bool # Human changed it? override_source: str = None # Who overrode? hold_duration: float # Seconds in PENDING notes: str = None # Human annotation ``` """, exports=["HoldPoint", "HoldState", "HoldResolution"], imports_from=["core.provenance"], imported_by=["hold.session", "viz.tape"], ), "hold.session": ModuleNode( name="session", full_path="cascade_lattice.hold.session", icon="๐ŸŽฎ", category="hold", parent="hold", doc_dummy=""" # ๐ŸŽฎ Hold Session **What is this?** The controller that manages HOLD pauses. It's like a game pause menu: - Press pause (yield_point) - Game freezes - You make a choice - Game resumes **Key actions:** - `yield_point()`: Pause and wait for human - `accept()`: Resume with AI's choice - `override()`: Resume with human's choice - `cancel()`: Abort entirely """, doc_scientist=""" # Hold Session (`cascade.hold.session`) Singleton-pattern HOLD controller with listener registration. ```python class Hold: _instance: 'Hold' = None # Singleton def __new__(cls) -> 'Hold': if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance def yield_point( self, action_probs: np.ndarray, value: float, observation: Any, brain_id: str, blocking: bool = True, **kwargs ) -> Optional[HoldResolution]: ''' Create HoldPoint, notify listeners, optionally block. Returns HoldResolution when resolved. ''' def accept(self, action: int = None) -> HoldResolution: ... def override(self, action: int, source: str) -> HoldResolution: ... def cancel(self) -> None: ... def register_listener(self, callback: Callable) -> None: ... def current_hold(self) -> Optional[HoldPoint]: ... @property def auto_accept(self) -> bool: ... @property def timeout(self) -> float: ... def stats(self) -> Dict[str, Any]: ... ``` ## Listener Pattern ```python def my_listener(hold_point: HoldPoint): print(f"Received: {hold_point.id}") # Inspect, then call Hold().accept() or override() Hold().register_listener(my_listener) ``` """, exports=["Hold", "HoldAwareMixin"], imports_from=["hold.primitives", "core.adapter"], imported_by=["store", "brain"], ), # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # STORE MODULE # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• "store": ModuleNode( name="store", full_path="cascade_lattice.store", icon="๐Ÿ’พ", category="store", parent="cascade_lattice", doc_dummy=""" # ๐Ÿ’พ Store **What is this?** Where all the receipts are saved. Like a filing cabinet that: - ๐Ÿ“ **Saves locally**: SQLite database on your computer - โ˜๏ธ **Syncs to cloud**: HuggingFace datasets - ๐ŸŒ **Pins to IPFS**: Decentralized, permanent storage **Key functions:** - `observe()`: Save an observation and get a receipt - `query()`: Find observations by model name - `stats()`: How many observations do I have? """, doc_scientist=""" # Store (`cascade.store`) Multi-tier persistence layer: Local SQLite โ†’ HuggingFace โ†’ IPFS. ```python # Configuration CENTRAL_DATASET = "tostido/cascade-observations" DEFAULT_LATTICE_DIR = Path.home() / ".cascade" / "lattice" IPFS_GATEWAYS = [ 'https://ipfs.io/ipfs/', 'https://dweb.link/ipfs/', 'https://gateway.pinata.cloud/ipfs/' ] @dataclass class Receipt: cid: str # IPFS CIDv1 (dag-cbor) model_id: str # Source model identifier merkle_root: str # Chain integrity hash timestamp: float # Unix epoch data: Dict[str, Any] # Original observation parent_cid: str = None class LocalStore: def __init__(self, lattice_dir: Path = None): ... def save(self, receipt: Receipt) -> None: ... def get(self, cid: str) -> Optional[Receipt]: ... def get_latest(self, model_id: str) -> Optional[Receipt]: ... def query(self, model_id: str, limit: int = 100) -> List[Receipt]: ... def count(self) -> int: ... # Top-level functions def observe(model_id: str, data: Dict, parent_cid: str = None, sync: bool = True) -> Receipt: ... def query(model_id: str, limit: int = 100) -> List[Receipt]: ... def stats() -> Dict[str, Any]: ... def sync_all() -> None: ... ``` ## CID Computation (dag-cbor) ```python def data_to_cid(data: Dict) -> Tuple[str, bytes]: cbor_bytes = dag_cbor.encode(data) digest = multihash.digest(cbor_bytes, 'sha2-256') return CID('base32', 1, 'dag-cbor', digest).encode(), cbor_bytes ``` """, exports=["Receipt", "LocalStore", "observe", "query", "stats", "sync_all", "compute_cid", "data_to_cid", "fetch_receipt"], imports_from=["core.provenance", "genesis"], imported_by=["hold.session", "viz"], ), # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # GENESIS MODULE # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• "genesis": ModuleNode( name="genesis", full_path="cascade_lattice.genesis", icon="๐ŸŒ…", category="genesis", parent="cascade_lattice", doc_dummy=""" # ๐ŸŒ… Genesis **What is this?** The beginning of everything. Like the first page of a book, or the Big Bang of the universe. Every single observation in cascade-lattice can trace its lineage back to ONE root hash: the Genesis Block. **The Genesis Message:** > "In the beginning was the hash, and the hash was with the chain, > and the hash was the chain." **Genesis Root:** `89f940c1a4b7aa65` This is your cryptographic anchor. If you can verify lineage to genesis, you KNOW the data hasn't been tampered with. """, doc_scientist=""" # Genesis (`cascade.genesis`) Root provenance anchor with deterministic initialization. ```python GENESIS_INPUT = "In the beginning was the hash, and the hash was with the chain, and the hash was the chain." GENESIS_MODEL_ID = "cascade_genesis" GENESIS_SESSION_ID = "genesis_0" def create_genesis() -> ProvenanceChain: ''' Create the root provenance chain. Deterministic: always produces merkle_root = 89f940c1a4b7aa65 ''' def get_genesis_root() -> str: '''Return the canonical genesis merkle root''' return "89f940c1a4b7aa65" def verify_lineage_to_genesis( chain: ProvenanceChain, known_chains: Dict[str, ProvenanceChain] ) -> bool: ''' Verify that a chain can trace its lineage to genesis. Walks external_roots until genesis is reached. ''' def link_to_genesis(chain: ProvenanceChain) -> ProvenanceChain: '''Add genesis as external root''' def save_genesis(path: Path) -> None: ... def load_genesis(path: Path) -> ProvenanceChain: ... ``` ## Genesis Chain Structure ```json { "model_id": "cascade_genesis", "session_id": "genesis_0", "merkle_root": "89f940c1a4b7aa65", "finalized": true, "records": [{ "input_hash": "", "output_hash": "", "model_hash": "", "merkle_root": "89f940c1a4b7aa65" }] } ``` """, exports=["create_genesis", "get_genesis_root", "verify_lineage_to_genesis", "link_to_genesis", "ProvenanceChain", "GENESIS_INPUT"], imports_from=["core.provenance"], imported_by=["store"], ), # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # VIZ MODULE # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• "viz": ModuleNode( name="viz", full_path="cascade_lattice.viz", icon="๐ŸŽฌ", category="viz", parent="cascade_lattice", doc_dummy=""" # ๐ŸŽฌ Viz - Visualization **What is this?** A VCR for AI decisions. Record everything that happens, then play it back later. Like watching game replays, but for AI behavior. **Features:** - ๐Ÿ“ผ **Tape**: Record events to a file - โช **Playback**: Scrub through history - ๐Ÿ” **Inspection**: See exactly what happened at any moment """, doc_scientist=""" # Viz (`cascade.viz`) Event tape recording and playback for debugging and analysis. ```python # Tape I/O def create_tape_path(prefix: str = "tape") -> str: '''Generate timestamped tape filename''' def write_tape_event(tape_path: str, event: Dict) -> None: '''Append JSONL event to tape file''' def load_tape_file(tape_path: str) -> List[Dict]: '''Load all events from tape''' def list_tape_files(directory: str = ".") -> List[str]: '''Find all tape files''' def find_latest_tape(directory: str = ".") -> Optional[str]: '''Get most recent tape file''' @dataclass class PlaybackEvent: timestamp: float event_type: str data: Dict[str, Any] @dataclass class PlaybackBuffer: events: List[PlaybackEvent] current_index: int = 0 is_complete: bool = False def add(self, event: PlaybackEvent) -> None: ... def get_events_up_to(self, index: int) -> List[PlaybackEvent]: ... @classmethod def from_tape(cls, tape_path: str) -> 'PlaybackBuffer': ... ``` """, exports=["PlaybackBuffer", "PlaybackEvent", "create_tape_path", "write_tape_event", "load_tape_file", "list_tape_files"], children=["viz.tape"], imports_from=["hold.primitives"], ), # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # DIAGNOSTICS MODULE # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• "diagnostics": ModuleNode( name="diagnostics", full_path="cascade_lattice.diagnostics", icon="๐Ÿ”ฌ", category="diagnostics", parent="cascade_lattice", doc_dummy=""" # ๐Ÿ”ฌ Diagnostics **What is this?** A doctor for your code. It scans your code looking for problems, bugs, and weird patterns. Like a health checkup, but for software. **Tools:** - ๐Ÿ› **BugDetector**: Find common bug patterns - ๐Ÿฅ **DiagnosticEngine**: Comprehensive code analysis - ๐Ÿ“Š **ExecutionMonitor**: Watch code as it runs """, doc_scientist=""" # Diagnostics (`cascade.diagnostics`) Static analysis, runtime monitoring, and bug pattern detection. ```python class BugDetector: def __init__(self): ... def register_pattern(self, pattern: BugPattern) -> None: ... def scan_file(self, path: Path) -> List[DetectedIssue]: ... def scan_directory(self, path: Path) -> List[DetectedIssue]: ... def get_summary(self) -> Dict[str, int]: ... def get_report(self) -> DiagnosticReport: ... class DiagnosticEngine: def analyze_file(self, path: Path) -> DiagnosticReport: ... def analyze_directory(self, path: Path) -> DiagnosticReport: ... def analyze_execution(self, trace: List[ExecutionFrame]) -> DiagnosticReport: ... def to_markdown(self) -> str: ... class ExecutionMonitor: def __init__(self): ... def start(self) -> None: ... def stop(self) -> List[ExecutionFrame]: ... def get_hot_paths(self) -> List[Tuple[str, int]]: ... @dataclass class BugPattern: name: str regex: str severity: str # info, warning, error, critical description: str fix_suggestion: str @dataclass class DetectedIssue: pattern: BugPattern file: Path line: int context: str ``` """, exports=["BugDetector", "DiagnosticEngine", "ExecutionMonitor", "BugPattern", "BugSignature", "DetectedIssue", "DiagnosticReport"], children=["diagnostics.bug_detector", "diagnostics.execution_monitor"], imports_from=["core.graph"], ), # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # FORENSICS MODULE # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• "forensics": ModuleNode( name="forensics", full_path="cascade_lattice.forensics", icon="๐Ÿ”Ž", category="forensics", parent="cascade_lattice", doc_dummy=""" # ๐Ÿ”Ž Forensics **What is this?** Digital archaeology for data. When you find mysterious data, forensics helps you figure out: - ๐Ÿ•ต๏ธ What created this data? - ๐Ÿ“… When was it created? - ๐Ÿ”ง What tools/frameworks were used? - ๐Ÿšจ Are there any security concerns? **Think of it like:** CSI for code and data. """, doc_scientist=""" # Forensics (`cascade.forensics`) Data archaeology and tech stack fingerprinting. ```python class DataForensics: def analyze(self, data: Any) -> ForensicsReport: ... def analyze_file(self, path: Path) -> ForensicsReport: ... class TechFingerprinter: PATTERNS: Dict[str, str] # tech_name โ†’ regex COMPOUND_PATTERNS: Dict[str, List[str]] # framework โ†’ components def analyze(self, text: str) -> List[Fingerprint]: ... def get_likely_stack(self) -> List[str]: ... def get_security_concerns(self) -> List[str]: ... class ArtifactDetector: def detect_timestamps(self, data: Any) -> TimestampArtifacts: ... def detect_ids(self, data: Any) -> IDPatternArtifacts: ... def detect_schemas(self, data: Any) -> SchemaArtifacts: ... def detect_text_patterns(self, text: str) -> TextArtifacts: ... def detect_numeric_patterns(self, numbers: List) -> NumericArtifacts: ... @dataclass class Fingerprint: tech: str confidence: float evidence: List[str] @dataclass class ForensicsReport: fingerprints: List[Fingerprint] artifacts: Dict[str, Any] inferred_operations: List[InferredOperation] ghost_logs: List[GhostLog] # Traces of deleted data ``` """, exports=["DataForensics", "TechFingerprinter", "ArtifactDetector", "Fingerprint", "ForensicsReport", "GhostLog"], children=["forensics.analyzer", "forensics.fingerprints", "forensics.artifacts"], ), # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # LISTEN MODULE # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• "listen": ModuleNode( name="listen", full_path="cascade_lattice.listen", icon="๐Ÿ‘‚", category="core", parent="cascade_lattice", doc_dummy=""" # ๐Ÿ‘‚ Listen **What is this?** A passive observer for HOLD events. Like having a security camera that records everything without interfering. Events flow through and you can watch them later. **The event_queue:** A line where events wait to be processed. """, doc_scientist=""" # Listen (`cascade.listen`) Passive event monitoring via thread-safe queue. ```python event_queue: Queue # Thread-safe event queue def main(): '''CLI entry point for passive monitoring''' monitor = Monitor("listener") while True: try: event = event_queue.get(timeout=1.0) print(json.dumps(event.to_dict())) except Empty: continue ``` Usage: ```bash python -m cascade.listen ``` """, exports=["event_queue", "Monitor"], imports_from=["core.event"], ), # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # MONITOR (top-level) # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• "monitor": ModuleNode( name="Monitor", full_path="cascade_lattice.Monitor", icon="๐Ÿ“ก", category="core", parent="cascade_lattice", doc_dummy=""" # ๐Ÿ“ก Monitor **What is this?** Your eyes and ears inside the system. It watches everything happening and can: - ๐Ÿ“ Record observations - ๐Ÿ” Trace cause and effect - ๐Ÿ”ฎ Predict what might happen next - ๐Ÿ•ต๏ธ Find root causes of problems **Like:** A flight recorder (black box) for AI systems. """, doc_scientist=""" # Monitor (`cascade_lattice.Monitor`) Unified observation and causation tracking facade. ```python class Monitor: def __init__(self, component: str): self.component = component self._graph = CausationGraph() self._adapter = SymbioticAdapter() def observe(self, signal: Any) -> Event: '''Interpret signal and add to causation graph''' event = self._adapter.interpret(signal) self._graph.add_event(event) return event def analyze_impact(self, event_id: str, max_depth: int = 20) -> ImpactReport: '''Forward trace: what did this event cause?''' def trace_backwards(self, event_id: str, max_depth: int = 10) -> List[CausationChain]: '''Backward trace: what caused this event?''' def trace_forwards(self, event_id: str, max_depth: int = 10) -> List[CausationChain]: '''Forward trace: what effects followed?''' def predict_cascade(self, event: Event) -> List[Event]: '''Predict likely downstream effects''' def find_root_causes(self, event_id: str) -> List[Event]: '''Find events with no causes in the subgraph''' ``` """, exports=["Monitor"], imports_from=["core.graph", "core.adapter", "core.event"], imported_by=["hold.session", "store"], ), } # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # NAVIGATION STATE # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• @dataclass class NavState: """Navigation breadcrumb trail.""" path: List[str] = field(default_factory=lambda: ["cascade_lattice"]) @property def current(self) -> str: return self.path[-1] if self.path else "cascade_lattice" def push(self, module: str) -> None: if module not in self.path: self.path.append(module) def pop(self) -> Optional[str]: if len(self.path) > 1: return self.path.pop() return None def jump_to(self, module: str) -> None: """Jump directly to a module (creative routing).""" if module in self.path: # Already visited - trim path to that point idx = self.path.index(module) self.path = self.path[:idx + 1] else: # New destination - add to path self.path.append(module) def reset(self) -> None: self.path = ["cascade_lattice"] # Global nav state _nav = NavState() # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # CUSTOM WIDGETS # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• class BreadcrumbWidget(Static): """Navigation breadcrumb trail.""" path = reactive(["cascade_lattice"]) def render(self) -> Text: text = Text() text.append("๐Ÿ“ ", style="bold") for i, module in enumerate(self.path): node = MODULE_GRAPH.get(module) icon = node.icon if node else "๐Ÿ“ฆ" if i > 0: text.append(" โ†’ ", style="dim") if i == len(self.path) - 1: # Current module (highlighted) text.append(f"{icon} {module}", style="bold cyan") else: # Previous modules (clickable look) text.append(f"{icon} {module}", style="blue underline") return text class ModuleCard(Static): """A clickable module card.""" def __init__(self, module_key: str, **kwargs): super().__init__(**kwargs) self.module_key = module_key self.node = MODULE_GRAPH.get(module_key, ModuleNode(name=module_key, full_path=module_key)) def compose(self) -> ComposeResult: yield Static(self._render_card(), id=f"card-content-{self.module_key}") def _render_card(self) -> Text: text = Text() text.append(f"{self.node.icon} ", style="bold") text.append(f"{self.node.name}\n", style="bold cyan") # Category badge cat_colors = { "root": "magenta", "core": "blue", "hold": "yellow", "store": "green", "genesis": "red", "viz": "cyan", "diagnostics": "orange1", "forensics": "purple", } color = cat_colors.get(self.node.category, "white") text.append(f"[{self.node.category}]", style=f"bold {color}") # Export count if self.node.exports: text.append(f" โ€ข {len(self.node.exports)} exports", style="dim") return text class RelatedModulesPanel(Vertical): """Shows related modules for creative navigation - now with clickable buttons!""" module_key = reactive("cascade_lattice") class ModuleClicked(Message): """Emitted when a related module is clicked.""" def __init__(self, module_key: str) -> None: self.module_key = module_key super().__init__() def compose(self) -> ComposeResult: yield Label("๐Ÿ”— CONNECTIONS", classes="panel-title") yield Vertical(id="related-buttons") def watch_module_key(self, module_key: str) -> None: """Rebuild buttons when module changes.""" self._rebuild_buttons() def on_mount(self) -> None: self._rebuild_buttons() def _rebuild_buttons(self) -> None: """Rebuild the related module buttons.""" try: container = self.query_one("#related-buttons", Vertical) container.remove_children() except NoMatches: return node = MODULE_GRAPH.get(self.module_key) if not node: return added_keys = set() # Track what we've added to avoid duplicates # Parent button if node.parent and node.parent not in added_keys: parent = MODULE_GRAPH.get(node.parent) if parent: btn = Button(f"โฌ†๏ธ {parent.icon} {parent.name}", variant="default", classes="related-btn") btn.tooltip = node.parent # Store module key in tooltip container.mount(btn) added_keys.add(node.parent) # Children buttons for child_key in node.children: if child_key not in added_keys: child = MODULE_GRAPH.get(child_key) if child: btn = Button(f"โฌ‡๏ธ {child.icon} {child.name}", variant="primary", classes="related-btn") btn.tooltip = child_key container.mount(btn) added_keys.add(child_key) # Imports from for imp in node.imports_from[:3]: if imp not in added_keys: imp_node = MODULE_GRAPH.get(imp) if imp_node: btn = Button(f"๐Ÿ“ฅ {imp_node.icon} {imp_node.name}", variant="success", classes="related-btn") btn.tooltip = imp container.mount(btn) added_keys.add(imp) # Imported by for imp in node.imported_by[:3]: if imp not in added_keys: imp_node = MODULE_GRAPH.get(imp) if imp_node: btn = Button(f"๐Ÿ“ค {imp_node.icon} {imp_node.name}", variant="warning", classes="related-btn") btn.tooltip = imp container.mount(btn) added_keys.add(imp) @on(Button.Pressed) def on_button_pressed(self, event: Button.Pressed) -> None: """Handle related module button click.""" # Get module key from button tooltip module_key = event.button.tooltip if module_key and module_key in MODULE_GRAPH: self.post_message(self.ModuleClicked(module_key)) class ExportsPanel(Static): """Shows exports of current module.""" module_key = reactive("cascade_lattice") def render(self) -> Text: node = MODULE_GRAPH.get(self.module_key) if not node or not node.exports: return Text("No exports") text = Text() text.append("๐Ÿ“ฆ EXPORTS\n\n", style="bold green") for exp in node.exports: # Guess type from name if exp[0].isupper(): text.append(" โ— ", style="cyan") text.append(f"{exp}\n", style="bold cyan") else: text.append(" โ—‹ ", style="yellow") text.append(f"{exp}()\n", style="yellow") return text class LiveStatsWidget(Static): """Live statistics from cascade-lattice store.""" def on_mount(self) -> None: self.set_interval(5.0, self._refresh_stats) self._refresh_stats() def _refresh_stats(self) -> None: try: from cascade_lattice.store import stats s = stats() self._stats = s except Exception as e: self._stats = {"error": str(e)} self.refresh() def render(self) -> Text: text = Text() text.append("๐Ÿ“Š LIVE STATS\n\n", style="bold magenta") if not hasattr(self, '_stats'): text.append("Loading...", style="dim") return text s = self._stats if "error" in s: text.append(f"Error: {s['error']}", style="red") return text text.append(f"Total Observations: ", style="dim") text.append(f"{s.get('total_observations', 0):,}\n", style="bold cyan") text.append(f"Pinned: ", style="dim") text.append(f"{s.get('pinned_observations', 0):,}\n", style="green") text.append(f"Genesis Root: ", style="dim") text.append(f"{s.get('genesis_root', 'unknown')}\n", style="bold yellow") models = s.get('models', {}) if models: text.append(f"\nTop Models:\n", style="dim") sorted_models = sorted(models.items(), key=lambda x: x[1], reverse=True)[:5] for name, count in sorted_models: short_name = name[:25] + "..." if len(name) > 25 else name text.append(f" {short_name}: ", style="white") text.append(f"{count:,}\n", style="cyan") return text # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # EXPLORER SCREEN # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• class ExplorerScreen(Screen): """Main module explorer with graph navigation.""" BINDINGS = [ Binding("escape", "go_back", "Back"), Binding("h", "go_home", "Home"), Binding("t", "toggle_mode", "Toggle Explanation"), Binding("s", "go_stats", "Stats"), Binding("d", "go_demo", "Demo"), ] # Explanation mode scientist_mode = reactive(False) current_module = reactive("cascade_lattice") def compose(self) -> ComposeResult: yield Header() with Container(id="explorer-main"): # Top bar with breadcrumb and toggle with Horizontal(id="top-bar"): yield BreadcrumbWidget(id="breadcrumb") yield Switch(value=False, id="mode-toggle") yield Label("๐Ÿงช Scientist Mode", id="mode-label") with Horizontal(id="content-area"): # Left: Module tree with Vertical(id="left-panel"): yield Label("๐Ÿ—‚๏ธ MODULES", classes="panel-title") yield Tree("cascade-lattice", id="module-tree") # Center: Documentation with Vertical(id="center-panel"): yield Label("๐Ÿ“– DOCUMENTATION", classes="panel-title") with ScrollableContainer(id="doc-scroll"): yield Markdown("", id="doc-content") # Right: Related + Exports with Vertical(id="right-panel"): yield RelatedModulesPanel(id="related-panel") yield Rule() yield ExportsPanel(id="exports-panel") yield Footer() def on_mount(self) -> None: self._build_tree() self._update_view() def _build_tree(self) -> None: """Build the module tree.""" tree = self.query_one("#module-tree", Tree) tree.clear() tree.root.expand() tree.show_root = True tree.guide_depth = 4 def add_children(parent_node: TreeNode, parent_key: str): node = MODULE_GRAPH.get(parent_key) if not node: return for child_key in node.children: child = MODULE_GRAPH.get(child_key) if child: child_node = parent_node.add( f"{child.icon} {child.name}", data=child_key ) add_children(child_node, child_key) # Start with root root = MODULE_GRAPH.get("cascade_lattice") if root: tree.root.set_label(f"{root.icon} cascade_lattice") tree.root.data = "cascade_lattice" add_children(tree.root, "cascade_lattice") def _update_view(self) -> None: """Update all panels for current module.""" node = MODULE_GRAPH.get(self.current_module) if not node: return # Update breadcrumb try: self.query_one("#breadcrumb", BreadcrumbWidget).path = list(_nav.path) except NoMatches: pass # Update documentation doc = node.doc_scientist if self.scientist_mode else node.doc_dummy try: self.query_one("#doc-content", Markdown).update(doc) except NoMatches: pass # Update related panel try: self.query_one("#related-panel", RelatedModulesPanel).module_key = self.current_module except NoMatches: pass # Update exports panel try: self.query_one("#exports-panel", ExportsPanel).module_key = self.current_module except NoMatches: pass @on(Tree.NodeSelected) def on_tree_select(self, event: Tree.NodeSelected) -> None: """Handle tree node selection.""" if event.node.data: _nav.jump_to(event.node.data) self.current_module = event.node.data self._update_view() @on(Switch.Changed, "#mode-toggle") def on_mode_toggle(self, event: Switch.Changed) -> None: """Toggle scientist/dummy mode.""" self.scientist_mode = event.value self._update_view() @on(RelatedModulesPanel.ModuleClicked) def on_related_clicked(self, event: RelatedModulesPanel.ModuleClicked) -> None: """Handle click on related module - creative navigation!""" _nav.jump_to(event.module_key) self.current_module = event.module_key self._update_view() # Also expand/select in tree self._select_in_tree(event.module_key) def _select_in_tree(self, module_key: str) -> None: """Find and select a module in the tree.""" try: tree = self.query_one("#module-tree", Tree) def find_node(node: TreeNode, key: str) -> Optional[TreeNode]: if node.data == key: return node for child in node.children: found = find_node(child, key) if found: return found return None target = find_node(tree.root, module_key) if target: target.expand() tree.select_node(target) except NoMatches: pass def watch_scientist_mode(self, value: bool) -> None: """React to mode change.""" try: label = self.query_one("#mode-label", Label) if value: label.update("๐Ÿงช Scientist Mode") else: label.update("๐Ÿ“š For Dummies") except NoMatches: pass def action_go_back(self) -> None: """Go back in navigation.""" popped = _nav.pop() if popped: self.current_module = _nav.current self._update_view() def action_go_home(self) -> None: """Go to root module.""" _nav.reset() self.current_module = "cascade_lattice" self._update_view() def action_toggle_mode(self) -> None: """Toggle explanation mode.""" toggle = self.query_one("#mode-toggle", Switch) toggle.value = not toggle.value def action_go_stats(self) -> None: """Switch to stats screen.""" self.app.switch_mode("stats") def action_go_demo(self) -> None: """Switch to demo screen.""" self.app.switch_mode("demo") # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # STATS SCREEN # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• class StatsScreen(Screen): """Live statistics dashboard.""" BINDINGS = [ Binding("escape", "go_explorer", "Explorer"), Binding("r", "refresh", "Refresh"), ] def compose(self) -> ComposeResult: yield Header() with Container(id="stats-main"): yield Label("๐Ÿ“Š CASCADE-LATTICE STATISTICS", classes="screen-title") with Horizontal(id="stats-row"): with Vertical(id="stats-left", classes="panel"): yield LiveStatsWidget(id="live-stats") with Vertical(id="stats-right", classes="panel"): yield Label("๐ŸŒ… GENESIS", classes="panel-title") yield Static(id="genesis-info") yield Rule() yield Label("๐Ÿ“ STORE", classes="panel-title") yield Static(id="store-info") yield Footer() def on_mount(self) -> None: self._load_genesis_info() self._load_store_info() def _load_genesis_info(self) -> None: try: import cascade_lattice as cl text = Text() text.append("Genesis Root: ", style="dim") text.append(f"{cl.genesis.get_genesis_root()}\n\n", style="bold yellow") text.append("Genesis Message:\n", style="dim") text.append(f'"{cl.genesis.GENESIS_INPUT}"', style="italic cyan") self.query_one("#genesis-info", Static).update(text) except Exception as e: self.query_one("#genesis-info", Static).update(f"Error: {e}") def _load_store_info(self) -> None: try: from cascade_lattice.store import DEFAULT_LATTICE_DIR, CENTRAL_DATASET text = Text() text.append("Local Store: ", style="dim") text.append(f"{DEFAULT_LATTICE_DIR}\n\n", style="cyan") text.append("Central Dataset: ", style="dim") text.append(f"{CENTRAL_DATASET}\n", style="green") self.query_one("#store-info", Static).update(text) except Exception as e: self.query_one("#store-info", Static).update(f"Error: {e}") def action_go_explorer(self) -> None: self.app.switch_mode("explorer") def action_refresh(self) -> None: self._load_genesis_info() self._load_store_info() self.notify("Refreshed") # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # DEMO SCREEN # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• class DemoScreen(Screen): """Interactive demo playground.""" BINDINGS = [ Binding("escape", "go_explorer", "Explorer"), Binding("1", "demo_hold", "HOLD Demo"), Binding("2", "demo_observe", "Observe Demo"), Binding("3", "demo_genesis", "Genesis Demo"), ] def compose(self) -> ComposeResult: yield Header() with Container(id="demo-main"): yield Label("๐ŸŽฎ INTERACTIVE DEMOS", classes="screen-title") with Horizontal(id="demo-buttons"): yield Button("โธ๏ธ HOLD Demo", id="btn-hold", variant="primary") yield Button("๐Ÿ‘๏ธ Observe Demo", id="btn-observe", variant="success") yield Button("๐ŸŒ… Genesis Demo", id="btn-genesis", variant="warning") yield Button("๐Ÿ” Provenance Demo", id="btn-provenance", variant="default") with Horizontal(id="demo-content"): with Vertical(id="demo-code", classes="panel"): yield Label("๐Ÿ“ CODE", classes="panel-title") yield Static(id="code-display") with Vertical(id="demo-output", classes="panel"): yield Label("๐Ÿ“ค OUTPUT", classes="panel-title") yield RichLog(id="output-log", max_lines=100) yield Footer() @on(Button.Pressed, "#btn-hold") def on_hold_demo(self) -> None: self._run_hold_demo() @on(Button.Pressed, "#btn-observe") def on_observe_demo(self) -> None: self._run_observe_demo() @on(Button.Pressed, "#btn-genesis") def on_genesis_demo(self) -> None: self._run_genesis_demo() @on(Button.Pressed, "#btn-provenance") def on_provenance_demo(self) -> None: self._run_provenance_demo() def _show_code(self, code: str) -> None: syntax = Syntax(code, "python", theme="monokai", line_numbers=True) self.query_one("#code-display", Static).update(syntax) def _log(self, message: str, style: str = "white") -> None: log = self.query_one("#output-log", RichLog) log.write(Text(message, style=style)) def _run_hold_demo(self) -> None: code = '''from cascade_lattice import Hold, HoldPoint, HoldState import numpy as np # Create a HoldPoint (freeze-frame snapshot) hp = HoldPoint( action_probs=np.array([0.1, 0.3, 0.6]), value=0.75, observation={"screen": "game_frame_42"}, brain_id="demo_agent" ) print(f"HoldPoint ID: {hp.id}") print(f"Merkle Root: {hp.merkle_root}") print(f"State: {hp.state}") print(f"Action Labels: Jump, Duck, Run")''' self._show_code(code) self._log("โ•โ•โ• HOLD DEMO โ•โ•โ•", "bold magenta") try: from cascade_lattice import Hold, HoldPoint, HoldState import numpy as np hp = HoldPoint( action_probs=np.array([0.1, 0.3, 0.6]), value=0.75, observation={"screen": "game_frame_42"}, brain_id="demo_agent", action_labels=["Jump", "Duck", "Run"] ) self._log(f"โœ“ Created HoldPoint", "green") self._log(f" ID: {hp.id}", "cyan") self._log(f" Merkle: {hp.merkle_root}", "yellow") self._log(f" State: {hp.state.value}", "white") self._log(f" Best Action: Run (60%)", "bold cyan") except Exception as e: self._log(f"โœ— Error: {e}", "red") def _run_observe_demo(self) -> None: code = '''from cascade_lattice.store import observe, query, stats # Create an observation receipt = observe( model_id="demo_model", data={"action": "explore", "reward": 1.0}, sync=False # Don't sync to HuggingFace ) print(f"CID: {receipt.cid}") print(f"Merkle: {receipt.merkle_root}") # Query recent observations recent = query("demo_model", limit=5) print(f"Found {len(recent)} observations")''' self._show_code(code) self._log("โ•โ•โ• OBSERVE DEMO โ•โ•โ•", "bold magenta") try: from cascade_lattice.store import observe, query import time receipt = observe( model_id="tui_demo", data={"action": "explore", "reward": 1.0, "timestamp": time.time()}, sync=False ) self._log(f"โœ“ Created Receipt", "green") self._log(f" CID: {receipt.cid[:40]}...", "cyan") self._log(f" Merkle: {receipt.merkle_root}", "yellow") recent = query("tui_demo", limit=5) self._log(f" Found {len(recent)} total observations", "white") except Exception as e: self._log(f"โœ— Error: {e}", "red") def _run_genesis_demo(self) -> None: code = '''import cascade_lattice as cl # Get the genesis root root = cl.genesis.get_genesis_root() print(f"Genesis Root: {root}") # The genesis message print(f"Message: {cl.genesis.GENESIS_INPUT}") # Create a provenance chain chain = cl.genesis.create_genesis() print(f"Chain verified: {chain.verify()}")''' self._show_code(code) self._log("โ•โ•โ• GENESIS DEMO โ•โ•โ•", "bold magenta") try: import cascade_lattice as cl root = cl.genesis.get_genesis_root() self._log(f"โœ“ Genesis Root: {root}", "bold yellow") self._log(f" \"{cl.genesis.GENESIS_INPUT}\"", "italic cyan") chain = cl.genesis.create_genesis() self._log(f" Chain Merkle: {chain.merkle_root}", "green") self._log(f" Finalized: {chain.finalized}", "white") except Exception as e: self._log(f"โœ— Error: {e}", "red") def _run_provenance_demo(self) -> None: code = '''from cascade_lattice.core.provenance import ( hash_tensor, hash_input, compute_merkle_root ) import numpy as np # Hash a tensor tensor = np.random.randn(100).astype(np.float32) t_hash = hash_tensor(tensor) # Hash an input input_data = {"query": "What is 2+2?", "context": []} i_hash = hash_input(input_data) # Compute merkle root root = compute_merkle_root([t_hash, i_hash]) print(f"Merkle Root: {root}")''' self._show_code(code) self._log("โ•โ•โ• PROVENANCE DEMO โ•โ•โ•", "bold magenta") try: from cascade_lattice.core.provenance import ( hash_tensor, hash_input, compute_merkle_root ) import numpy as np tensor = np.random.randn(100).astype(np.float32) t_hash = hash_tensor(tensor) self._log(f"โœ“ Tensor hash: {t_hash}", "cyan") input_data = {"query": "What is 2+2?", "context": []} i_hash = hash_input(input_data) self._log(f"โœ“ Input hash: {i_hash}", "yellow") root = compute_merkle_root([t_hash, i_hash]) self._log(f"โœ“ Merkle root: {root}", "bold green") except Exception as e: self._log(f"โœ— Error: {e}", "red") def action_go_explorer(self) -> None: self.app.switch_mode("explorer") def action_demo_hold(self) -> None: self._run_hold_demo() def action_demo_observe(self) -> None: self._run_observe_demo() def action_demo_genesis(self) -> None: self._run_genesis_demo() # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # MAIN APP # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• class CascadeLatticeTUI(App): """The Cascade-Lattice Explorer TUI.""" TITLE = "๐ŸŒ CASCADE-LATTICE" SUB_TITLE = "even still, i grow, and yet, I grow still" CSS = """ Screen { background: $surface; } #explorer-main, #stats-main, #demo-main { layout: vertical; padding: 1; } #top-bar { height: 3; padding: 0 1; align: left middle; background: $surface-darken-1; } #breadcrumb { width: 1fr; } #mode-toggle { width: auto; margin-right: 1; } #mode-label { width: auto; } #content-area { height: 1fr; } #left-panel { width: 25%; min-width: 30; border: solid $primary; padding: 1; } #center-panel { width: 50%; border: solid $secondary; padding: 1; } #right-panel { width: 25%; min-width: 25; border: solid $accent; padding: 1; } #doc-scroll { height: 1fr; } .panel-title { text-style: bold; color: $text; padding-bottom: 1; } .screen-title { text-style: bold; text-align: center; color: $primary; padding: 1; } #module-tree { height: 1fr; scrollbar-gutter: stable; } .related-btn { width: 100%; margin: 0 0 1 0; } #related-buttons { height: auto; padding: 1 0; } /* Stats screen */ #stats-row { height: 1fr; } #stats-left, #stats-right { width: 1fr; border: solid $primary; padding: 1; margin: 1; } /* Demo screen */ #demo-buttons { height: auto; padding: 1; align: center middle; } #demo-buttons Button { margin: 0 1; } #demo-content { height: 1fr; } #demo-code, #demo-output { width: 1fr; border: solid $primary; padding: 1; margin: 1; } #code-display { height: 1fr; } #output-log { height: 1fr; } .panel { border: solid $surface-lighten-1; padding: 1; margin: 0 1; } """ BINDINGS = [ Binding("e", "go_explorer", "Explorer", show=True), Binding("s", "go_stats", "Stats", show=True), Binding("d", "go_demo", "Demo", show=True), Binding("q", "quit", "Quit", show=True), ] MODES = { "explorer": ExplorerScreen, "stats": StatsScreen, "demo": DemoScreen, } def on_mount(self) -> None: """Start on explorer.""" self.switch_mode("explorer") def action_go_explorer(self) -> None: self.switch_mode("explorer") def action_go_stats(self) -> None: self.switch_mode("stats") def action_go_demo(self) -> None: self.switch_mode("demo") # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # ENTRY POINT # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• def main(): """Run the TUI.""" app = CascadeLatticeTUI() app.run() if __name__ == "__main__": main()