Add dynamic terminal execution block to simulation.py

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()
-

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.

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()

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