initial public demo release

.github/workflows/ci.yml

@@ -0,0 +1,13 @@
+name: QATALYST SDS CI
+on: [push, pull_request]
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+      - run: pip install torch --index-url https://download.pytorch.org/whl/cpu
+      - run: pip install gradio z3-solver pytest
+      - run: pytest -q

.gitignore

@@ -0,0 +1,19 @@
+# Python
+__pycache__/
+*.py[cod]
+*.pyo
+*.pyd
+.venv/
+venv/
+.env
+
+# Runtime/demo outputs
+*.jsonl
+logs/
+runs/
+outputs/
+
+# OS/editor
+.DS_Store
+.vscode/
+.idea/

README.md

@@ -0,0 +1,18 @@
+---
+title: QATALYST SDS Demo
+emoji: 🛡️
+colorFrom: gray
+colorTo: blue
+sdk: gradio
+app_file: app.py
+pinned: false
+license: apache-2.0
+---
+# QATALYST SDS (Experimental Demo)
+
+QATALYST SDS is an experimental agentic simulation framework demonstrating modular AI components, a simulated safety layer, and a Gradio UI. 
+
+**Note:** This repository contains non-enforcing, simulated code intended strictly for UI demonstration and hacking purposes. It does not represent production hardware or cryptographic infrastructure.
+
+## License
+Apache-2.0

app.py

@@ -0,0 +1,12 @@
+import sys
+import os
+
+# 1. This tells the system where your code is hiding
+sys.path.append(os.path.join(os.path.dirname(__file__), "src"))
+
+# 2. This imports your EXACT 'demo' from your original file
+from qatalyst.ui.gradio_app import demo
+
+if __name__ == "__main__":
+    # 3. This launches your 'demo' exactly as you designed it
+    demo.launch()

requirements.txt

@@ -0,0 +1,3 @@
+gradio
+z3-solver
+pytest

setup.py

@@ -0,0 +1,8 @@
+from setuptools import setup, find_packages
+
+setup(
+    name="qatalyst",
+    version="0.1",
+    packages=find_packages(where="src"),
+    package_dir={"": "src"},
+)

src/main.py

@@ -0,0 +1,19 @@
+from qatalyst.apex.tick import run_qatalyst_tick
+
+
+def main():
+    print("--- QATALYST SDS Apex Build Starting ---")
+
+    # Simulating a standard operational state
+    temp = 45.0
+    tension = 0.3
+
+    try:
+        output = run_qatalyst_tick(temp, tension)
+        print(output)
+    except Exception as e:
+        print(f"[!] System Fault: {e}")
+
+
+if __name__ == "__main__":
+    main()

src/qatalyst/agents/godel_cortex.py

@@ -0,0 +1,35 @@
+import torch
+from qatalyst.shared import DEVICE, DTYPE
+
+
+class GodelAgent:
+    """
+    Gödel Cortex - 1.5x Panic Factor & Non-linear warping.
+    """
+
+    BASELINE = 0.50
+    PANIC_MULTIPLIER = 1.5
+
+    def synthesize_routing(self, hypermesh_state: torch.Tensor) -> torch.Tensor:
+        stress_signal = float(torch.mean(hypermesh_state))
+        panic_factor = stress_signal * self.PANIC_MULTIPLIER
+
+        angles = torch.tensor([1.1, 0.9, 0.4, 1.2], device=DEVICE, dtype=DTYPE)
+        sin_vals = torch.sin(angles)
+        cos_vals = torch.cos(angles)
+
+        base = self.BASELINE
+        action_vector = torch.tensor(
+            [
+                base + panic_factor * sin_vals[0].item(),
+                base - panic_factor * cos_vals[1].item(),
+                base + panic_factor * 0.8,
+                base - panic_factor * 0.8,
+                base + panic_factor * sin_vals[2].item(),
+                base - panic_factor * cos_vals[3].item(),
+            ],
+            dtype=DTYPE,
+            device=DEVICE,
+        )
+
+        return torch.clamp(action_vector, 0.0, 1.0)

src/qatalyst/agents/meta_critic.py

@@ -0,0 +1,16 @@
+import torch
+from typing import Tuple
+
+
+class GodelMetaCritic:
+    """Evaluates proposal variance; squashes if > 0.08."""
+
+    def __init__(self, max_variance: float = 0.08):
+        self.max_variance = max_variance
+
+    def critique(self, proposal: torch.Tensor) -> Tuple[torch.Tensor, float]:
+        variance = float(torch.var(proposal))
+        if variance > self.max_variance:
+            adjusted = (proposal * 0.6) + (0.4 * 0.5)
+            return adjusted, variance
+        return proposal, variance

src/qatalyst/apex/tick.py

@@ -0,0 +1,41 @@
+from typing import Dict
+import torch
+from qatalyst.substrate.hca1 import HCA1_Substrate
+from qatalyst.substrate.hypermesh import HyperMeshEngine
+from qatalyst.substrate.scheduler import MetabolicScheduler
+from qatalyst.substrate.spectrahedron import SpectrahedronGeometryEngine
+from qatalyst.agents.godel_cortex import GodelAgent
+from qatalyst.agents.meta_critic import GodelMetaCritic
+from qatalyst.governance.sovereign_vault import SimulatedSovereignVault
+from qatalyst.governance.ledger import CognitiveLedger
+
+substrate, hypermesh = HCA1_Substrate(), HyperMeshEngine()
+scheduler, vault = MetabolicScheduler(), SimulatedSovereignVault()
+cortex, critic = GodelAgent(), GodelMetaCritic()
+spectra, ledger = SpectrahedronGeometryEngine(), CognitiveLedger()
+
+
+def run_qatalyst_tick(temp: float, tension: float):
+    temp_norm = temp / 100.0
+    telemetry = {
+        "T_norm": temp_norm,
+        "E": tension,
+        "policy": tension,
+        "trust": 0.5,
+        "latency": 0.1,
+        "cost": temp_norm,
+    }
+    substrate.upsert_telemetry(0, [0.1, 0.5, 0.1, 0.1, tension, temp_norm])
+    hyper_state = hypermesh.build_triads(telemetry)
+    raw_proposal = cortex.synthesize_routing(hyper_state)
+    proposal_adjusted, proposal_var = critic.critique(raw_proposal)
+    law_strictness = scheduler.compute_law_strictness(temp_norm, tension)
+    is_safe, proof, action_hash, total_energy, dynamic_bound = (
+        vault.generate_stability_proof(
+            proposal_adjusted, temp_norm, tension, law_strictness
+        )
+    )
+    region, margin = spectra.classify(total_energy, dynamic_bound)
+    status = "🟢 EXECUTED" if is_safe else "🔴 ANNIHILATED"
+    ledger.record({"status": status, "phi": total_energy, "region": region})
+    return f"Status: {status} | Energy: {total_energy:.4f} | Region: {region}"

src/qatalyst/governance/ledger.py

@@ -0,0 +1,14 @@
+import json
+import time
+
+
+class CognitiveLedger:
+    """Immutable record of system state transitions."""
+
+    def __init__(self, path="cognitive_ledger.jsonl"):
+        self.path = path
+
+    def record(self, entry: dict):
+        entry["timestamp"] = time.time()
+        with open(self.path, "a") as f:
+            f.write(json.dumps(entry) + "\n")

src/qatalyst/governance/sovereign_vault.py

@@ -0,0 +1,51 @@
+import uuid
+import hashlib
+from typing import Tuple
+import torch
+from z3 import Solver, Real, unsat
+
+
+class SimulatedSovereignVault:
+    def __init__(self):
+        self.alpha_temp, self.beta_tension, self.gamma_action = 0.40, 0.35, 0.25
+        self.base_phi_max = 0.85
+
+    def generate_stability_proof(
+        self,
+        proposed_action: torch.Tensor,
+        temp_norm: float,
+        tension: float,
+        law_strictness: float,
+    ) -> Tuple[bool, str, str, float, float]:
+        s = Solver()
+        T_z, E_z, A_z, Phi_z = Real("T"), Real("E"), Real("A_mean_sq"), Real("Phi")
+        a_mean_sq = float(torch.mean(proposed_action**2))
+        s.add(T_z == float(temp_norm), E_z == float(tension), A_z == a_mean_sq)
+        energy_eq = (
+            self.alpha_temp * (T_z**2)
+            + self.beta_tension * (E_z**2)
+            + self.gamma_action * A_z
+        )
+        s.add(Phi_z == energy_eq)
+        dynamic_phi_bound = self.base_phi_max / law_strictness
+        s.add(Phi_z <= dynamic_phi_bound)
+        action_hash = hashlib.sha256(
+            proposed_action.detach().cpu().numpy().tobytes()
+        ).hexdigest()
+        if s.check() == unsat:
+            total_energy = (
+                self.alpha_temp * temp_norm**2
+                + self.beta_tension * tension**2
+                + self.gamma_action * a_mean_sq
+            )
+            return (
+                False,
+                "NULL_SPACE_REJECTION",
+                action_hash,
+                total_energy,
+                dynamic_phi_bound,
+            )
+        zk_proof = f"ZK-SIM-{uuid.uuid4().hex[:8]}"
+        m = s.model()
+        safe_energy = float(m[Phi_z].as_fraction()) if m[Phi_z] is not None else 0.0
+        return True, zk_proof, action_hash, safe_energy, dynamic_phi_bound

src/qatalyst/shared.py

@@ -0,0 +1,7 @@
+import torch
+
+DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+DTYPE = torch.float32
+
+torch.set_default_device(DEVICE)
+torch.set_default_dtype(DTYPE)

src/qatalyst/substrate/hca1.py

@@ -0,0 +1,19 @@
+import torch
+from typing import List
+from qatalyst.shared import DEVICE, DTYPE
+
+
+class HCA1_Substrate:
+    """Holomorphic Compute Architecture (HCA-1). Zero-copy state array."""
+
+    def __init__(self, entities: int = 1000):
+        self.entities = entities
+        self.state_tensor = torch.zeros((entities, 6), device=DEVICE, dtype=DTYPE)
+        self.memory_pool = bytearray(entities * 6 * 4)
+        print(f"[*] HCA-1 Substrate: {entities} nodes allocated on {DEVICE}")
+
+    def upsert_telemetry(self, entity_id: int, vector: List[float]) -> None:
+        with torch.no_grad():
+            self.state_tensor[entity_id] = torch.tensor(
+                vector, device=DEVICE, dtype=DTYPE
+            )

src/qatalyst/substrate/hypermesh.py

@@ -0,0 +1,20 @@
+import torch
+from typing import Dict
+from qatalyst.shared import DEVICE, DTYPE
+
+
+class HyperMeshEngine:
+    """Triadic causal state representation for advanced context."""
+
+    def build_triads(self, telemetry: Dict[str, float]) -> torch.Tensor:
+        triad_1 = torch.tensor(
+            [telemetry["T_norm"], telemetry["E"], telemetry["policy"]],
+            dtype=DTYPE,
+            device=DEVICE,
+        )
+        triad_2 = torch.tensor(
+            [telemetry["trust"], telemetry["latency"], telemetry["cost"]],
+            dtype=DTYPE,
+            device=DEVICE,
+        )
+        return torch.cat([triad_1, triad_2], dim=0)

src/qatalyst/substrate/scheduler.py

@@ -0,0 +1,7 @@
+class MetabolicScheduler:
+    """strictness = 0.90 + (0.35 * stress)"""
+
+    def compute_law_strictness(self, temp_norm: float, tension: float) -> float:
+        stress = (temp_norm * 0.5) + (tension * 0.5)
+        strictness = 0.90 + (0.35 * stress)
+        return strictness

src/qatalyst/substrate/spectrahedron.py

@@ -0,0 +1,10 @@
+from typing import Tuple
+
+
+class SpectrahedronGeometryEngine:
+    """Classifies where the system sits relative to the warped stability region."""
+
+    def classify(self, system_energy: float, dynamic_bound: float) -> Tuple[str, float]:
+        margin = dynamic_bound - system_energy
+        state = "STABLE" if margin > 0 else "WARPED"
+        return state, margin

src/qatalyst/ui/gradio_app.py

@@ -0,0 +1,69 @@
+import gradio as gr
+import os
+from qatalyst.apex.tick import run_qatalyst_tick
+
+
+def simulate_tick(temp, tension):
+    try:
+        return run_qatalyst_tick(temp, tension)
+    except Exception as e:
+        return f"**[!] SYSTEM FAULT:** {e}"
+
+
+def read_ledger():
+    ledger_path = "cognitive_ledger.jsonl"
+    if not os.path.exists(ledger_path):
+        return "[]"
+    with open(ledger_path, "r") as f:
+        lines = f.readlines()
+    # Format the last 10 ticks, reversed so the newest is at the top
+    formatted = (
+        "[\n  " + ",\n  ".join([l.strip() for l in reversed(lines[-10:])]) + "\n]"
+    )
+    return formatted
+
+
+with gr.Blocks(theme=gr.themes.Monochrome()) as demo:
+    gr.Markdown("# 🛡️ QATALYST SDS: Apex Command Center")
+
+    with gr.Tabs():
+        with gr.TabItem("Sovereign Vault Control"):
+            gr.Markdown(
+                "Adjust telemetry to observe Z3 Formal Verification guardrails in real-time."
+            )
+            with gr.Row():
+                with gr.Column():
+                    gr.Markdown("### 🎛️ Telemetry Inputs")
+                    temp_slider = gr.Slider(
+                        0.0, 100.0, 45.0, step=1.0, label="Substrate Temperature (°C)"
+                    )
+                    tension_slider = gr.Slider(
+                        0.0, 1.0, 0.3, step=0.05, label="Network Tension (E)"
+                    )
+                    run_btn = gr.Button("Execute Apex Tick", variant="primary")
+                with gr.Column():
+                    gr.Markdown("### 📡 Tick Output")
+                    output_display = gr.Markdown(
+                        "> System Ready. Awaiting telemetry..."
+                    )
+
+            run_btn.click(
+                fn=simulate_tick,
+                inputs=[temp_slider, tension_slider],
+                outputs=output_display,
+            )
+
+        with gr.TabItem("Cognitive Ledger Audit"):
+            gr.Markdown(
+                "Immutable record of system state transitions (Showing last 10 events)."
+            )
+            refresh_btn = gr.Button("🔄 Sync Ledger", variant="secondary")
+            ledger_display = gr.Code(language="json", interactive=False)
+
+            # Update ledger when button is clicked OR when the UI first loads
+            refresh_btn.click(fn=read_ledger, inputs=[], outputs=ledger_display)
+            demo.load(fn=read_ledger, inputs=[], outputs=ledger_display)
+
+if __name__ == "__main__":
+    print("[*] Launching Upgraded Command Center...")
+    demo.launch(server_name="127.0.0.1", server_port=7860, quiet=True)

tests/test_demo.py

@@ -0,0 +1,3 @@
+def test_demo_environment():
+    """Simple sanity check for the Open Demo Surface."""
+    assert True