diff --git a/.coverage b/.coverage index e3c694f..ce566f5 100644 Binary files a/.coverage and b/.coverage differ diff --git a/examples/run_simulation.py b/examples/run_simulation.py index cf09cf0..5cb91be 100755 --- a/examples/run_simulation.py +++ b/examples/run_simulation.py @@ -14,9 +14,13 @@ CostWeights, StabilityBandConfig ) - +from radial_membrane_ai.ufo_engine.serialization import ( + export_simulation_results_to_json, + ledger_to_json +) from radial_membrane_ai.exceptions import GovernanceError + def main() -> None: print("=" * 70) print("🛸 U.F.O. GOVERNED SIMULATION ENGINE DEMO 🛸") @@ -158,4 +162,3 @@ def main() -> None: if __name__ == "__main__": main() - diff --git a/logs/governance_halts.log b/logs/governance_halts.log index 261a36e..b6b4a05 100644 --- a/logs/governance_halts.log +++ b/logs/governance_halts.log @@ -1,4 +1,2 @@ -[2026-07-21 01:59:35] Test Halt Message -[2026-07-21 02:01:36] GovernanceError correctly raised during local simulation due to temporal tension exceeding hard limit. Confirms stability enforcement and deterministic halt behavior. -[2026-07-20 22:23:19] GovernanceError correctly raised during local simulation due to temporal tension exceeding hard limit. Confirms stability enforcement and deterministic halt behavior. -[2026-07-20 22:37:35] GovernanceError correctly raised during local simulation due to temporal tension exceeding hard limit. Confirms stability enforcement and deterministic halt behavior. +[2026-07-21 03:00:42] Test Halt Message +[2026-07-21 03:01:26] GovernanceError correctly raised during local simulation due to temporal tension exceeding hard limit. Confirms stability enforcement and deterministic halt behavior. diff --git a/logs/visualization/ufo_simulation_timeline.png b/logs/visualization/ufo_simulation_timeline.png index 89367a9..3e454da 100644 Binary files a/logs/visualization/ufo_simulation_timeline.png and b/logs/visualization/ufo_simulation_timeline.png differ diff --git a/radial_membrane_ai/simulation.py b/radial_membrane_ai/simulation.py index 62e6d1d..d4fa49b 100644 --- a/radial_membrane_ai/simulation.py +++ b/radial_membrane_ai/simulation.py @@ -26,6 +26,11 @@ from radial_membrane_ai.envelope import BrimEnvelope from radial_membrane_ai.saopromotion import SAOPromotor from radial_membrane_ai.mesh import FederatedShardMesh, FederatedShard, ShardState +from radial_membrane_ai.utils import set_deterministic_env +from radial_membrane_ai.exceptions import GovernanceError +import sys +import time +import os class RainbowSimulation: @@ -428,3 +433,106 @@ def render_summary(self) -> None: for q in ["analytical", "contextual", "generative", "interpersonal"]: print(f" {q.capitalize()}: {self.membrane.get_quadrant_activation(q):.4f}") print("======================================") + + +# Color formatting constants +ANSI_PURPLE = "\033[95m" +ANSI_BLUE = "\033[94m" +ANSI_GREEN = "\033[92m" +ANSI_YELLOW = "\033[93m" +ANSI_RED = "\033[91m" +ANSI_CYAN = "\033[96m" +ANSI_BOLD = "\033[1m" +ANSI_RESET = "\033[0m" + + +def print_banner(text: str, color: str = ANSI_CYAN) -> None: + border = "=" * 75 + print(f"{color}{border}") + print(f"{ANSI_BOLD}{text.center(75)}{ANSI_RESET}{color}") + print(f"{border}{ANSI_RESET}\n") + + +def run_governed_simulation() -> None: + """ + Runs a deterministic, 10-tick governed simulation using RainbowSimulation. + Supports animated progression, color-coded metrics, and robust GovernanceError handling. + """ + # 1. Deterministic Seeding Confirmation + set_deterministic_env(0) + print(f"{ANSI_GREEN}{ANSI_BOLD}[Deterministic Seed Initialized: 0]{ANSI_RESET}\n") + + # 2. Cinematic Banner + print_banner("🛸 UFO SIMULATION MODULE EXECUTING 🛸", ANSI_PURPLE) + + # 3. Instantiate + sim = RainbowSimulation() + + steps = 1 if os.getenv("UFO_TEST_RUN") else 10 + task_sequence = [ + "technical_deep_analysis", + "supportive_concise_reply", + "planning" + ] * 4 # Cycles through available tasks (at least 12 steps, sliced to 10) + + try: + for step_idx in range(steps): + tick_num = step_idx + 1 + + # Animated tick prefix + anim_chars = ["◐", "◓", "◑", "◒"] + anim = anim_chars[step_idx % len(anim_chars)] + print(f"{ANSI_BOLD}{ANSI_CYAN}{anim} Tick {tick_num}/{steps}...{ANSI_RESET}", end="\r") + sys.stdout.flush() + time.sleep(0.05) + + # Run step + sim.run_step(task_sequence[step_idx]) + + # Extract and format metrics + curv = max([sim.boundary.curvature(s.theta) for s in sim.membrane.strings]) + t_state = getattr(sim.membrane, "temporal_state", None) + tens = t_state.accumulated_tension if t_state is not None else 0.0 + coh = sim.mesh_coherence_history[-1] if sim.mesh_coherence_history else 1.0 + energy = sim.governor.energy_history[-1] if sim.governor.energy_history else 0.0 + + if energy <= 1.0: + band = "GREEN" + band_color = ANSI_GREEN + elif energy <= 2.0: + band = "YELLOW" + band_color = ANSI_YELLOW + else: + band = "RED" + band_color = ANSI_RED + + # Color format strings + curv_str = f"curvature: {ANSI_PURPLE}{curv:.4f}{ANSI_RESET}" + tens_str = f"tension: {ANSI_BLUE}{tens:.4f}{ANSI_RESET}" + coh_str = f"coherence: {ANSI_GREEN}{coh:.4f}{ANSI_RESET}" + band_str = f"stability band: {band_color}{band}{ANSI_RESET}" + + tick_p = f"[{ANSI_BOLD}Tick {tick_num:2d}/{steps:2d}{ANSI_RESET}]" + print(f"{tick_p} - {curv_str} | {tens_str} | {coh_str} | {band_str} ") + + print() # newline after steps complete + sim.render_summary() + + except GovernanceError as e: + print("\n" + "=" * 75) + print(f"{ANSI_RED}{ANSI_BOLD}🚨 GOVERNED HALT ENFORCED 🚨{ANSI_RESET}") + print("=" * 75) + print(f"{ANSI_RED}{ANSI_BOLD}Violation Details:{ANSI_RESET} {e}") + print( + f"{ANSI_YELLOW}{ANSI_BOLD}Halt Ledger Code:{ANSI_RESET} " + "GovernanceError raised during simulation execution." + ) + print("=" * 75 + "\n") + sim.render_summary() + + print_banner("🛸 MODULE EXECUTION COMPLETE — RETURNING TO SHELL 🛸", ANSI_PURPLE) + input("Press Enter to exit...") + + +if __name__ == "__main__": + run_governed_simulation() diff --git a/radial_membrane_ai/tests/test_simulation_main.py b/radial_membrane_ai/tests/test_simulation_main.py new file mode 100644 index 0000000..d31defc --- /dev/null +++ b/radial_membrane_ai/tests/test_simulation_main.py @@ -0,0 +1,116 @@ +""" +Tests for the dynamic terminal execution block of radial_membrane_ai/simulation.py. +""" + +from __future__ import annotations +import sys +import os +import subprocess +from unittest.mock import patch, MagicMock + +from radial_membrane_ai.exceptions import GovernanceError +from radial_membrane_ai.simulation import run_governed_simulation, print_banner + + +def test_print_banner() -> None: + # Verify that printing a banner executes without failure + with patch("builtins.print") as mock_print: + print_banner("Test Banner Text") + mock_print.assert_called() + + +def test_run_governed_simulation_nominal() -> None: + # Test nominal execution of the simulation run wrapper + # We patch time.sleep to run instantly and input to prevent hanging. + with patch("time.sleep"), \ + patch("builtins.input", return_value=""), \ + patch("builtins.print") as mock_print: + run_governed_simulation() + + # Verify deterministic initialization, cinematic banners, ticks, and completion printed + printed_texts = [call[0][0] for call in mock_print.call_args_list if call[0]] + assert any("[Deterministic Seed Initialized: 0]" in text for text in printed_texts) + assert any("🛸 UFO SIMULATION MODULE EXECUTING 🛸" in text for text in printed_texts) + assert any("🛸 MODULE EXECUTION COMPLETE — RETURNING TO SHELL 🛸" in text for text in printed_texts) + + +def test_run_governed_simulation_red_band() -> None: + # Test that the RED stability band branch is taken when energy is high + sim_instance = MagicMock() + # Mock governor energy_history to return > 2.0 + sim_instance.governor.energy_history = [3.0] + sim_instance.mesh_coherence_history = [0.8] + mock_string = MagicMock() + mock_string.theta = 0.0 + sim_instance.membrane.strings = [mock_string] + sim_instance.boundary.curvature.return_value = 1.5 + # Configure the temporal_state mock to return a float for accumulated_tension + sim_instance.membrane.temporal_state.accumulated_tension = 0.5 + + with patch("time.sleep"), \ + patch("builtins.input", return_value=""), \ + patch("radial_membrane_ai.simulation.RainbowSimulation", return_value=sim_instance), \ + patch("builtins.print") as mock_print: + run_governed_simulation() + + printed_texts = [call[0][0] for call in mock_print.call_args_list if call[0]] + # Check that RED stability band was printed + assert any("RED" in text for text in printed_texts) + + +def test_run_governed_simulation_halt() -> None: + # Test execution when a GovernanceError is raised + # We mock run_step on RainbowSimulation to raise GovernanceError + with patch("time.sleep"), \ + patch("builtins.input", return_value=""), \ + patch("radial_membrane_ai.simulation.RainbowSimulation") as mock_sim_class, \ + patch("builtins.print") as mock_print: + + # Setup mock instance of RainbowSimulation to raise GovernanceError during run_step + mock_sim_inst = MagicMock() + mock_sim_inst.run_step.side_effect = GovernanceError("Simulated governance breach of temporal limits.") + mock_sim_class.return_value = mock_sim_inst + + run_governed_simulation() + + printed_texts = [call[0][0] for call in mock_print.call_args_list if call[0]] + assert any("🚨 GOVERNED HALT ENFORCED 🚨" in text for text in printed_texts) + assert any("Violation Details:" in text for text in printed_texts) + assert any("Simulated governance breach of temporal limits." in text for text in printed_texts) + mock_sim_inst.render_summary.assert_called() + + +def test_simulation_main_direct() -> None: + # Read simulation.py and execute it with __name__ set to "__main__" + # to get 100% test coverage including the __main__ block + with open("radial_membrane_ai/simulation.py") as f: + code = f.read() + + with patch("time.sleep"), \ + patch("builtins.input", return_value=""), \ + patch("builtins.print"): + # Create a dictionary for execution globals + globals_dict = {"__name__": "__main__"} + # Execute the module code + exec(code, globals_dict) + + +def test_main_subprocess() -> None: + # Run python -m radial_membrane_ai.simulation as a subprocess to hit the __main__ block + # We pass an empty newline to standard input to simulate pressing Enter. + # We pass UFO_TEST_RUN=1 to run a single step quickly. + env = dict(os.environ) + env["UFO_TEST_RUN"] = "1" + + p = subprocess.Popen( + [sys.executable, "-m", "radial_membrane_ai.simulation"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=env, + text=True + ) + stdout, stderr = p.communicate(input="\n", timeout=15) + assert p.returncode == 0 + assert "🛸 UFO SIMULATION MODULE EXECUTING 🛸" in stdout + assert "🛸 MODULE EXECUTION COMPLETE — RETURNING TO SHELL 🛸" in stdout