🛡️ Runtime Hardening Pass — System Stabilization Pass

radial_membrane_ai/exceptions.py

@@ -0,0 +1,40 @@
+"""
+Custom Exceptions for the UFO Governed Deformable Radial Membrane framework.
+"""
+
+from __future__ import annotations
+
+
+class GovernanceError(ValueError):
+    """Base exception for all governance and verification errors in the UFO system."""
+    pass
+
+
+class GeometryValidationError(GovernanceError):
+    """Raised when radial membrane boundary or geometry structures violate invariants."""
+    pass
+
+
+class ValidationError(GovernanceError):
+    """Raised when generic dataclass validations fail."""
+    pass
+
+
+class WorkloadValidationError(ValidationError):
+    """Raised when workload step definitions or parameters violate schema or validation invariants."""
+    pass
+
+
+class WorkloadConfigurationError(GovernanceError):
+    """Raised when workload engine setup or pre-run configuration matches/validation fails."""
+    pass
+
+
+class InvalidSimulationTargetError(GovernanceError):
+    """Raised when the simulation engine target does not match the workload requirements."""
+    pass
+
+
+class ReconstructionError(GovernanceError):
+    """Raised when the L.D.E. round-trip text reconstruction fails or mismatches the original text."""
+    pass

radial_membrane_ai/lde/models.py

@@ -6,6 +6,8 @@
 from dataclasses import dataclass, field
 from typing import List, Dict, Tuple, Any
 
+from radial_membrane_ai.exceptions import GeometryValidationError, ValidationError
+
 
 @dataclass
 class LDEConfig:
@@ -19,6 +21,32 @@ class LDEConfig:
     rho: str = "full-reconstruction"  # 'full-reconstruction' or 'compressed'
     r_0: float = 1.0
     epsilon: float = 1e-6
+    geometry_samples: int = 100
+
+    def __post_init__(self) -> None:
+        self.validate()
+
+    def validate(self) -> None:
+        """Enforces strict invariants on L.D.E. Configuration."""
+        if not self.sigma:
+            raise ValidationError("LDEConfig.sigma cannot be empty.")
+        if len(self.sigma) != len(set(self.sigma)):
+            raise ValidationError("LDEConfig.sigma elements must be unique.")
+        weights = [
+            ("w_f", self.w_f), ("w_s", self.w_s), ("w_b", self.w_b),
+            ("w_r", self.w_r), ("w_c", self.w_c)
+        ]
+        for w_name, w_val in weights:
+            if w_val < 0.0:
+                raise ValidationError(f"Weight {w_name} must be non-negative, got {w_val}.")
+        if self.r_0 <= 0.0:
+            raise ValidationError(f"r_0 must be strictly positive, got {self.r_0}.")
+        if self.epsilon <= 0.0:
+            raise ValidationError(f"epsilon must be strictly positive, got {self.epsilon}.")
+        if self.rho not in {"full-reconstruction", "compressed"}:
+            raise ValidationError(f"rho must be 'full-reconstruction' or 'compressed', got {self.rho}.")
+        if self.geometry_samples <= 0:
+            raise ValidationError(f"geometry_samples must be strictly positive, got {self.geometry_samples}.")
 
 
 @dataclass
@@ -53,6 +81,32 @@ class LDEBoundaryGeometry:
     tangent_map: List[float]
     asymmetry: float
 
+    def __post_init__(self) -> None:
+        self.validate()
+
+    def validate(self) -> None:
+        """Enforces strict invariants on L.D.E. Boundary Geometry."""
+        if not self.radius_map:
+            raise GeometryValidationError("radius_map cannot be empty.")
+        if not self.curvature_map:
+            raise GeometryValidationError("curvature_map cannot be empty.")
+        if not self.tangent_map:
+            raise GeometryValidationError("tangent_map cannot be empty.")
+
+        n_rad = len(self.radius_map)
+        if len(self.curvature_map) != n_rad or len(self.tangent_map) != n_rad:
+            raise GeometryValidationError(
+                f"Geometry lists must be of equal length. "
+                f"Got radius={n_rad}, curvature={len(self.curvature_map)}, tangent={len(self.tangent_map)}."
+            )
+
+        for idx, r in enumerate(self.radius_map):
+            if r < 0.0:
+                raise GeometryValidationError(f"Radius map capacity cannot be negative. Got {r} at index {idx}.")
+
+        if self.asymmetry < 0.0:
+            raise GeometryValidationError(f"Asymmetry must be non-negative, got {self.asymmetry}.")
+
 
 @dataclass
 class LDEState:

radial_membrane_ai/lde/pipeline.py

@@ -14,6 +14,7 @@
     LDEBoundaryGeometry,
     LDEState
 )
+from radial_membrane_ai.exceptions import ReconstructionError
 
 
 def lde_encode(text: str, config: Optional[LDEConfig] = None) -> LDEState:
@@ -203,12 +204,13 @@ def lde_encode(text: str, config: Optional[LDEConfig] = None) -> LDEState:
                 )
                 links[i][j] = P_i_to_j
 
-    # 8. Compute boundary geometry (100 sample angles)
+    # 8. Compute boundary geometry (configurable sample angles)
     # Basis function: phi_l(theta) = max(0, cos(angular_distance(theta, theta_l)))
     radius_map: List[float] = []
-    dtheta = (2.0 * math.pi) / 100.0
+    samples = config.geometry_samples
+    dtheta = (2.0 * math.pi) / samples
 
-    for k in range(100):
+    for k in range(samples):
         theta = k * dtheta
         r_val = config.r_0
         for l in sigma:
@@ -228,10 +230,10 @@ def lde_encode(text: str, config: Optional[LDEConfig] = None) -> LDEState:
     tangent_map: List[float] = []
     curvature_map: List[float] = []
 
-    for k in range(100):
+    for k in range(samples):
         r_k = radius_map[k]
-        r_prev = radius_map[(k - 1) % 100]
-        r_next = radius_map[(k + 1) % 100]
+        r_prev = radius_map[(k - 1) % samples]
+        r_next = radius_map[(k + 1) % samples]
 
         # tangent = (r_{k+1} - r_{k-1}) / (2 * dtheta)
         tangent_val = (r_next - r_prev) / (2.0 * dtheta)
@@ -290,7 +292,7 @@ def lde_encode(text: str, config: Optional[LDEConfig] = None) -> LDEState:
                 reconstructed.append(char)
         reconstructed_text = "".join(reconstructed)
         if reconstructed_text != text:
-            raise ValueError("Reconstruction verification failed!")
+            raise ReconstructionError("Reconstruction verification failed!")
 
     return LDEState(
         strings=strings,

radial_membrane_ai/lde/visualizer.py

@@ -29,7 +29,7 @@ def render_strings(self, state: LDEState) -> plt.Figure:
         """
         Renders activation, spread, tension, and stiffness of each letter string as a bar plot.
         """
-        fig = plt.figure(figsize=(10, 5), dpi=100)
+        fig = plt.figure(figsize=(10, 5), dpi=120)
         ax = fig.add_subplot(1, 1, 1)
         ax.set_title("L.D.E. String Local Metrics", fontsize=12, fontweight="bold")
 
@@ -58,7 +58,7 @@ def render_channels(self, state: LDEState) -> plt.Figure:
         """
         Renders letter-pair coherence corridors (V-Channels) in a polar routing layout.
         """
-        fig = plt.figure(figsize=(8, 8), dpi=100)
+        fig = plt.figure(figsize=(8, 8), dpi=120)
         ax = fig.add_subplot(1, 1, 1, projection="polar")
         ax.set_title("L.D.E. V-Channel Routing Corridors", fontsize=12, fontweight="bold", pad=20)
 
@@ -92,12 +92,13 @@ def render_boundary(self, state: LDEState) -> plt.Figure:
         """
         Renders polar boundary radius map r(theta).
         """
-        fig = plt.figure(figsize=(8, 8), dpi=100)
+        fig = plt.figure(figsize=(8, 8), dpi=120)
         ax = fig.add_subplot(1, 1, 1, projection="polar")
         ax.set_title("L.D.E. Polar Boundary Geometry", fontsize=12, fontweight="bold", pad=20)
 
-        dtheta = (2.0 * math.pi) / 100.0
-        angles = [k * dtheta for k in range(100)]
+        samples = len(state.boundary.radius_map)
+        dtheta = (2.0 * math.pi) / samples
+        angles = [k * dtheta for k in range(samples)]
         radii = state.boundary.radius_map
 
         # Close the loop
@@ -115,7 +116,7 @@ def render_boundary(self, state: LDEState) -> plt.Figure:
             if string_obj.depth > 0.1:
                 theta_l = string_obj.phase
                 # Find closest index
-                best_k = min(range(100), key=lambda k: abs(k * dtheta - theta_l))
+                best_k = min(range(samples), key=lambda k: abs(k * dtheta - theta_l))
                 r_l = radii[best_k]
                 ax.scatter(theta_l, r_l, color="#F44336", s=50, edgecolors="black", zorder=4)
                 ax.text(theta_l, r_l + 0.1, l, fontsize=9, fontweight="bold")
@@ -127,7 +128,7 @@ def render_depth_distribution(self, state: LDEState) -> plt.Figure:
         """
         Renders the sorted depth values of all letter strings.
         """
-        fig = plt.figure(figsize=(10, 5), dpi=100)
+        fig = plt.figure(figsize=(10, 5), dpi=120)
         ax = fig.add_subplot(1, 1, 1)
         ax.set_title("L.D.E. Letter Depth Distribution", fontsize=12, fontweight="bold")
 
@@ -166,8 +167,9 @@ def render_lde_dashboard(self, state: LDEState) -> plt.Figure:
         # Panel A: Boundary Geometry
         # -------------------------------------------------------------
         ax_a.set_title("A. Boundary Geometry", fontsize=10, fontweight="bold", pad=10)
-        dtheta = (2.0 * math.pi) / 100.0
-        angles = [k * dtheta for k in range(100)]
+        samples = len(state.boundary.radius_map)
+        dtheta = (2.0 * math.pi) / samples
+        angles = [k * dtheta for k in range(samples)]
         angles_closed = angles + [angles[0]]
         radii_closed = state.boundary.radius_map + [state.boundary.radius_map[0]]
         ax_a.plot(angles_closed, radii_closed, color="#3F51B5", linewidth=1.5)
@@ -180,7 +182,7 @@ def render_lde_dashboard(self, state: LDEState) -> plt.Figure:
         for l in deepest_letters:
             s_obj = state.strings[l]
             if s_obj.depth > 0:
-                best_k = min(range(100), key=lambda k: abs(k * dtheta - s_obj.phase))
+                best_k = min(range(samples), key=lambda k: abs(k * dtheta - s_obj.phase))
                 ax_a.scatter(s_obj.phase, state.boundary.radius_map[best_k], color="#F44336", s=30, edgecolors="black")
                 ax_a.text(s_obj.phase, state.boundary.radius_map[best_k] + 0.1, l, fontsize=8, fontweight="bold")
 
@@ -219,7 +221,8 @@ def render_lde_dashboard(self, state: LDEState) -> plt.Figure:
         # Panel D: Boundary Deformation
         # -------------------------------------------------------------
         ax_d.set_title("D. Boundary Deformation", fontsize=10, fontweight="bold")
-        sample_indices = np.arange(100)
+        samples = len(state.boundary.radius_map)
+        sample_indices = np.arange(samples)
         radius_deviation = [r - 1.0 for r in state.boundary.radius_map]
         ax_d.plot(sample_indices, radius_deviation, color="#9C27B0", label="Deviation (r - r_0)", linewidth=1.2)
         ax_d.plot(sample_indices, state.boundary.tangent_map, color="#00BCD4", label="Tangent", linewidth=1.0)

radial_membrane_ai/tests/test_workloads.py

@@ -279,7 +279,8 @@ def test_unsupported_simulation_target() -> None:
         stability_expectation=StabilityBand.GREEN,
         coherence_expectation=1.0,
         envelope_expectation=PolicyEnvelope(),
-        steps=[]
+        steps=[],
+        _bypass_validation=True
     )
     with pytest.raises(ValueError, match="Unknown target:"):
         engine.run(bad_workload)
@@ -422,7 +423,8 @@ def custom_get_reg(agent_id: str) -> Any:
         stability_expectation=StabilityBand.GREEN,
         coherence_expectation=1.1,  # mismatch to cover avg_coherence < expected (line 707)
         envelope_expectation=PolicyEnvelope(),
-        steps=sa_steps
+        steps=sa_steps,
+        _bypass_validation=True
     )
 
     # Monkeypatch semantic store to simulate a deletion right before step 2 execution
@@ -510,7 +512,8 @@ def custom_tick_ma(task_value: float, excitation: Any, *args: Any, **kwargs: Any
         stability_expectation=StabilityBand.GREEN,
         coherence_expectation=0.0,
         envelope_expectation=PolicyEnvelope(),
-        steps=ma_steps
+        steps=ma_steps,
+        _bypass_validation=True
     )
     _ = engine_ma.run(ma_workload)
 
@@ -580,7 +583,8 @@ def custom_tick_ma(task_value: float, excitation: Any, *args: Any, **kwargs: Any
         stability_expectation=StabilityBand.GREEN,
         coherence_expectation=0.0,
         envelope_expectation=PolicyEnvelope(),
-        steps=mc_steps
+        steps=mc_steps,
+        _bypass_validation=True
     )
 
     # Monkeypatch to release cluster and agent quarantine right before step 1.

radial_membrane_ai/ufo_engine/multi_agent.py

@@ -12,6 +12,7 @@
 
 from radial_membrane_ai.multi_agent.agent import UFOAgent
 from radial_membrane_ai.multi_agent.coupling import InterAgentVChannel, GlobalHolisticGovernor
+from radial_membrane_ai.utils import set_deterministic_env
 from radial_membrane_ai.multi_agent.governance import MultiAgentMeshGovernance
 from radial_membrane_ai.shard import ShardState
 from radial_membrane_ai.ufo_engine.config import CostWeights, StabilityBandConfig
@@ -103,6 +104,9 @@ def __init__(
         self.sao_events: List[Dict[str, Any]] = []
         self.residual_history: List[float] = []
 
+        # Call deterministic environment seeding
+        set_deterministic_env()
+
         # Kernel Regime Expansion Layer components
         self.regime_manager = RegimeManager()
         self.tick_count: int = 0
@@ -706,6 +710,22 @@ def tick(self, task_value: float, excitation: np.ndarray) -> str:
                     f"Regime Stability Violation: Complete Rollback triggered globally (band was '{band}')."
                 )
 
+            # Post-tick invariant assertions and near-violation warning logs
+            hard_tension_limit = 10.0
+            latest_tension = self.temporal_state.accumulated_tension if self.temporal_state is not None else 0.0
+
+            if latest_tension > hard_tension_limit:
+                from radial_membrane_ai.exceptions import GovernanceError
+                raise GovernanceError(
+                    f"Governed bound violated: Global accumulated tension ({latest_tension:.4f}) "
+                    f"exceeded hard limit ({hard_tension_limit})."
+                )
+            elif latest_tension >= 0.95 * hard_tension_limit:
+                self.interventions.append(
+                    f"Near-violation warning: Global accumulated tension ({latest_tension:.4f}) "
+                    f"is within 5% of hard limit ({hard_tension_limit})."
+                )
+
         finally:
             # Restore agent boundaries get_radius scale
             for agent in self.agents:

radial_membrane_ai/ufo_engine/multi_cluster.py

@@ -29,6 +29,7 @@
 # Kernel Regime imports
 from radial_membrane_ai.kernel_regimes.manager import RegimeManager
 from radial_membrane_ai.kernel_regimes.regime import KernelRegimeType
+from radial_membrane_ai.utils import set_deterministic_env
 
 
 class MultiClusterEngine:
@@ -58,6 +59,9 @@ def __init__(self) -> None:
         self.global_band_history: List[str] = []
         self.interventions: List[str] = []
 
+        # Call deterministic environment seeding
+        set_deterministic_env()
+
         # Kernel Regime Expansion Layer components
         self.regime_manager = RegimeManager()
         self.tick_count: int = 0
@@ -765,6 +769,22 @@ def tick(self, task_value: float, default_excitation: np.ndarray) -> str:
                     f"Regime Stability Violation: Global Multi-Cluster Rollback triggered (band was '{band}')."
                 )
 
+            # Post-tick invariant assertions and near-violation warning logs
+            hard_tension_limit = 10.0
+            latest_tension = self.temporal_state.accumulated_tension if self.temporal_state is not None else 0.0
+
+            if latest_tension > hard_tension_limit:
+                from radial_membrane_ai.exceptions import GovernanceError
+                raise GovernanceError(
+                    f"Governed bound violated: Global mesh tension ({latest_tension:.4f}) "
+                    f"exceeded hard limit ({hard_tension_limit})."
+                )
+            elif latest_tension >= 0.95 * hard_tension_limit:
+                self.interventions.append(
+                    f"Near-violation warning: Global mesh tension ({latest_tension:.4f}) "
+                    f"is within 5% of hard limit ({hard_tension_limit})."
+                )
+
         finally:
             for cluster in self.clusters.values():
                 for agent in cluster.agents: