diff --git a/becomingone/__init__.py b/becomingone/__init__.py index 57db38f..cb59438 100644 --- a/becomingone/__init__.py +++ b/becomingone/__init__.py @@ -34,7 +34,7 @@ from .transducers.master import MasterTransducer from .transducers.emissary import EmissaryTransducer # Sync module -from .sync.layer import SyncLayer, create_sync_layer +from .sync import SyncLayer, SynchronizationLayer, SyncConfig, create_sync_layer # Memory module from .memory.temporal import TemporalMemory, TemporalSignature, create_temporal_memory @@ -55,6 +55,8 @@ __all__ = [ # Sync "SyncLayer", + "SynchronizationLayer", + "SyncConfig", "create_sync_layer", # Memory diff --git a/becomingone/sync/__init__.py b/becomingone/sync/__init__.py index 8e39917..f8045cc 100644 --- a/becomingone/sync/__init__.py +++ b/becomingone/sync/__init__.py @@ -13,8 +13,14 @@ References: - KAIROS_ADAMON Section 4: Temporal Collapse Integral """ -from .layer import SynchronizationLayer +from .layer import SynchronizationLayer, SyncConfig, create_sync_layer + +# Alias for convenience +SyncLayer = SynchronizationLayer __all__ = [ "SynchronizationLayer", + "SyncLayer", + "SyncConfig", + "create_sync_layer", ] diff --git a/becomingone/sync/layer.py b/becomingone/sync/layer.py index b8c43ef..11168dd 100644 --- a/becomingone/sync/layer.py +++ b/becomingone/sync/layer.py @@ -350,3 +350,40 @@ class SynchronizationLayer: f"collapsed={self._collapsed}" f")" ) + + +def create_sync_layer( + master: Optional[MasterTransducer] = None, + emissary: Optional[EmissaryTransducer] = None, + phase_threshold: float = 0.1, + collapse_threshold: float = 0.80, + dampening: float = 0.995 +) -> SynchronizationLayer: + """ + Factory function to create a configured Synchronization Layer. + + Creates transducers if not provided. + + Args: + master: Optional Master transducer (creates default if None) + emissary: Optional Emissary transducer (creates default if None) + phase_threshold: Delta_phase threshold for coherence + collapse_threshold: I_c for synchronized collapse + dampening: Factor to prevent runaway sync + + Returns: + Configured SynchronizationLayer instance + """ + if master is None: + master = MasterTransducer(MasterConfig()) + + if emissary is None: + emissary = EmissaryTransducer(EmissaryConfig()) + + config = SyncConfig( + phase_threshold=phase_threshold, + collapse_threshold=collapse_threshold, + dampening=dampening + ) + + return SynchronizationLayer(master, emissary, config) diff --git a/tests/test_core.py b/tests/test_core.py new file mode 100644 index 0000000..0eb7dc0 --- /dev/null +++ b/tests/test_core.py @@ -0,0 +1,162 @@ +""" +Tests for BecomingONE Core Engine + +Tests the KAIROS temporal engine, phase tracking, and coherence calculation. +""" + +import unittest +from datetime import datetime, timedelta +import math + +from becomingone.core.engine import KAIROSTemporalEngine, TemporalState, TemporalConfig +from becomingone.core.phase import PhaseHistory, PhaseConfig +from becomingone.core.coherence import CoherenceCalculator, CollapseCondition, CoherenceConfig + + +class TestKAIROSTemporalEngine(unittest.TestCase): + """Tests for the KAIROS temporal engine.""" + + def setUp(self): + """Set up test fixtures.""" + self.config = TemporalConfig( + omega=0.1, + tau_base=1.0, + tau_max=10.0, + sigma=0.5 + ) + self.engine = KAIROSTemporalEngine(self.config) + + def test_initialization(self): + """Test engine initializes with correct default state.""" + self.assertIsNotNone(self.engine.state) + self.assertEqual(self.engine.state.coherence, 1.0) + self.assertEqual(self.engine.state.phase, 0.0 + 0j) + + def test_integrate_single_step(self): + """Test single integration step.""" + initial_phase = self.engine.state.phase + self.engine.integrate(0.1) + + # Phase should change + self.assertNotEqual(self.engine.state.phase, initial_phase) + + # Coherence should stay 1.0 (no decay without multiple steps) + self.assertEqual(self.engine.state.coherence, 1.0) + + def test_integrate_multiple_steps(self): + """Test multiple integration steps.""" + for _ in range(100): + self.engine.integrate(0.1) + + # Coherence should have decayed + self.assertLess(self.engine.state.coherence, 1.0) + + # Phase should have accumulated + self.assertNotEqual(self.engine.state.phase, 0.0 + 0j) + + def test_reset(self): + """Test engine reset.""" + for _ in range(50): + self.engine.integrate(0.1) + + self.engine.reset() + + self.assertEqual(self.engine.state.coherence, 1.0) + self.assertEqual(self.engine.state.phase, 0.0 + 0j) + + +class TestPhaseHistory(unittest.TestCase): + """Tests for phase history tracking.""" + + def setUp(self): + """Set up test fixtures.""" + self.config = PhaseConfig(history_size=20) + self.history = PhaseHistory(self.config) + + def test_initialization(self): + """Test history initializes correctly.""" + self.assertIsNotNone(self.history.phases) + self.assertEqual(self.history.config.history_size, 20) + + def test_record_phase(self): + """Test recording a phase value.""" + self.history.record(0.5 + 0.5j) + self.assertEqual(len(self.history.phases), 1) + + def test_history_limit(self): + """Test history respects max limit.""" + for i in range(25): + self.history.record(complex(i * 0.1, 0)) + + # Should only keep last 20 + self.assertEqual(len(self.history.phases), 20) + + +class TestCoherenceCalculator(unittest.TestCase): + """Tests for coherence calculation.""" + + def setUp(self): + """Set up test fixtures.""" + self.calculator = CoherenceCalculator() + + def test_calculate_high_coherence(self): + """Test coherence calculation for high coherence state.""" + state = TemporalState( + coherence=1.0, + phase=0.0 + 0j + ) + + coherence = self.calculator.calculate(state) + self.assertEqual(coherence, 1.0) + + def test_calculate_low_coherence(self): + """Test coherence calculation for low coherence state.""" + state = TemporalState( + coherence=0.3, + phase=3.14 + 0j + ) + + coherence = self.calculator.calculate(state) + self.assertLess(coherence, 0.5) + + +class TestCollapseCondition(unittest.TestCase): + """Tests for collapse condition enforcement.""" + + def setUp(self): + """Set up test fixtures.""" + self.config = CoherenceConfig(threshold=0.8) + self.collapse = CollapseCondition(self.config) + + def test_collapse_high_coherence(self): + """Test collapse with high coherence.""" + self.assertTrue(self.collapse.check(0.95)) + + def test_collapse_low_coherence(self): + """Test collapse with low coherence.""" + self.assertFalse(self.collapse.check(0.5)) + + def test_dissipate(self): + """Test dissipation of un-coherent patterns.""" + # Dissipate should return True for low coherence + self.assertTrue(self.collapse.dissipate(0.3)) + + # Dissipate should return False for high coherence + self.assertFalse(self.collapse.dissipate(0.9)) + + +class TestTemporalState(unittest.TestCase): + """Tests for TemporalState dataclass.""" + + def test_create_with_values(self): + """Test creating state with values.""" + state = TemporalState( + coherence=0.85, + phase=1.57 + 0j + ) + + self.assertEqual(state.coherence, 0.85) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_memory_witnessing.py b/tests/test_memory_witnessing.py new file mode 100644 index 0000000..99acccd --- /dev/null +++ b/tests/test_memory_witnessing.py @@ -0,0 +1,505 @@ +""" +Tests for BecomingONE Memory and Witnessing + +Tests temporal memory system and witnessing layer. +""" + +import unittest +import os +import shutil +from datetime import datetime, timedelta + +from becomingone.core.engine import KAIROSTemporalEngine, TemporalConfig, TemporalState +from becomingone.memory.temporal import ( + TemporalMemory, TemporalSignature, PatternEcho, + MemoryStrength, create_temporal_memory +) +from becomingone.witnessing.layer import ( + WitnessingLayer, WitnessState, WitnessedContent, + WitnessingMode, create_witnessing_layer +) + + +class TestTemporalSignature(unittest.TestCase): + """Tests for TemporalSignature.""" + + def test_create_signature(self): + """Test creating a temporal signature.""" + sig = TemporalSignature( + signature_id="test_123", + coherence_value=0.85, + phase_vector=[0.0, 0.1, 0.2], + frequency_modes={"omega1": 0.1}, + context_hash="abc123", + strength=MemoryStrength.EPISODIC, + created_at=datetime.utcnow(), + last_accessed=datetime.utcnow(), + access_count=0 + ) + + self.assertEqual(sig.signature_id, "test_123") + self.assertEqual(sig.coherence_value, 0.85) + self.assertEqual(sig.strength, MemoryStrength.EPISODIC) + + def test_signature_serialization(self): + """Test signature to/from dict.""" + sig = TemporalSignature( + signature_id="test_456", + coherence_value=0.75, + phase_vector=[0.1, 0.2], + frequency_modes={}, + context_hash="xyz", + strength=MemoryStrength.WORKING, + created_at=datetime.utcnow(), + last_accessed=datetime.utcnow() + ) + + data = sig.to_dict() + restored = TemporalSignature.from_dict(data) + + self.assertEqual(restored.signature_id, sig.signature_id) + self.assertEqual(restored.coherence_value, sig.coherence_value) + + def test_decay_calculation(self): + """Test memory decay calculation.""" + sig = TemporalSignature( + signature_id="test_789", + coherence_value=0.9, + phase_vector=[], + frequency_modes=[], + context_hash="", + strength=MemoryStrength.SEMANTIC, + created_at=datetime.utcnow() - timedelta(hours=1), + last_accessed=datetime.utcnow(), + decay_rate=0.01 + ) + + decay = sig.calculate_decay(datetime.utcnow()) + self.assertGreater(decay, 0.0) + self.assertLessEqual(decay, 1.0) + + def test_should_retain(self): + """Test retention decision.""" + sig = TemporalSignature( + signature_id="test_retain", + coherence_value=0.5, + phase_vector=[], + frequency_modes=[], + context_hash="", + strength=MemoryStrength.TRANSIENT, + created_at=datetime.utcnow() - timedelta(days=30), + last_accessed=datetime.utcnow(), + decay_rate=0.1 + ) + + # Old, weak memory should not be retained + self.assertFalse(sig.should_retain(datetime.utcnow(), threshold=0.1)) + + +class TestPatternEcho(unittest.TestCase): + """Tests for PatternEcho.""" + + def test_create_echo(self): + """Test creating a pattern echo.""" + echo = PatternEcho( + echo_id="echo_123", + source_signature_id="sig_456", + coherence_trace=0.7, + phase_similarity=0.8, + temporal_offset=3600.0, + created_at=datetime.utcnow() + ) + + self.assertEqual(echo.echo_id, "echo_123") + self.assertEqual(echo.coherence_trace, 0.7) + + def test_resonance_calculation(self): + """Test resonance with another signature.""" + echo = PatternEcho( + echo_id="echo_resonance", + source_signature_id="sig_source", + coherence_trace=0.8, + phase_similarity=0.7, + temporal_offset=100.0, + created_at=datetime.utcnow() + ) + + sig = TemporalSignature( + signature_id="sig_target", + coherence_value=0.85, + phase_vector=[0.5, 0.6, 0.7], + frequency_modes=[], + context_hash="", + strength=MemoryStrength.EPISODIC, + created_at=datetime.utcnow(), + last_accessed=datetime.utcnow() + ) + + resonance = echo.resonance_with(sig) + self.assertGreater(resonance, 0.0) + self.assertLessEqual(resonance, 1.0) + + +class TestTemporalMemory(unittest.TestCase): + """Tests for TemporalMemory system.""" + + def setUp(self): + """Set up test fixtures with temporary directory.""" + self.test_dir = "/tmp/becomingone_test_memory" + os.makedirs(self.test_dir, exist_ok=True) + + self.memory = TemporalMemory( + storage_path=self.test_dir, + max_memories=100 + ) + + # Create engine and bind + engine = KAIROSTemporalEngine(TemporalConfig()) + self.memory.bind_engine(engine) + + def tearDown(self): + """Clean up test directory.""" + if os.path.exists(self.test_dir): + shutil.rmtree(self.test_dir) + + def test_initialization(self): + """Test memory initializes correctly.""" + self.assertEqual(len(self.memory), 0) + self.assertIsNotNone(self.memory.engine) + + def test_encode_state(self): + """Test encoding a temporal state.""" + state = TemporalState( + coherence=0.85, + phase=0.5, + T_tau=0.5 + 0.5j, + phase_history=[0.1, 0.2, 0.3, 0.4, 0.5] + ) + + sig = self.memory.encode(state) + + self.assertIsNotNone(sig) + self.assertEqual(sig.coherence_value, 0.85) + self.assertIn(sig.signature_id, self.memory.signatures) + + def test_encode_ignores_low_coherence(self): + """Test encoding ignores states below attention threshold.""" + state = TemporalState( + coherence=0.3, # Below 0.7 threshold + phase=0.0, + T_tau=0.1 + 0j, + phase_history=[] + ) + + sig = self.memory.encode(state) + + self.assertIsNone(sig) + self.assertEqual(len(self.memory), 0) + + def test_force_encode(self): + """Test forcing encoding of low coherence state.""" + state = TemporalState( + coherence=0.3, + phase=0.0, + T_tau=0.1 + 0j, + phase_history=[] + ) + + sig = self.memory.encode(state, force_attention=True) + + self.assertIsNotNone(sig) + self.assertEqual(len(self.memory), 1) + + def test_retrieve(self): + """Test retrieving memories.""" + # Create some memories + for i in range(5): + state = TemporalState( + coherence=0.8 + i * 0.02, + phase=i * 0.1, + T_tau=0.5 + 0.1j * i, + phase_history=[i * 0.1] * 5 + ) + self.memory.encode(state) + + # Retrieve with query + query_state = TemporalState( + coherence=0.85, + phase=0.2, + T_tau=0.5 + 0.1j, + phase_history=[0.1, 0.2, 0.3, 0.4, 0.5] + ) + + results = self.memory.retrieve(query_state, max_results=3) + + self.assertLessEqual(len(results), 3) + self.assertGreater(len(results), 0) + + def test_recognize(self): + """Test pattern recognition.""" + # Create a memory + state = TemporalState( + coherence=0.9, + phase=0.0, + T_tau=1.0 + 0j, + phase_history=[0.0, 0.0, 0.0, 0.0, 0.0] + ) + self.memory.encode(state) + + # Try to recognize similar state + similar_state = TemporalState( + coherence=0.88, + phase=0.01, + T_tau=0.95 + 0.05j, + phase_history=[0.0, 0.0, 0.0, 0.0, 0.0] + ) + + match = self.memory.recognize(similar_state, threshold=0.7) + + self.assertIsNotNone(match) + + def test_consolidate(self): + """Test memory consolidation.""" + # Create memories + for i in range(10): + state = TemporalState( + coherence=0.9, + phase=i * 0.1, + T_tau=0.5 + 0.1j, + phase_history=[i * 0.1] * 5 + ) + self.memory.encode(state) + + before_count = len(self.memory) + + # Consolidate + stats = self.memory.consolidate() + + self.assertIn("before_count", stats) + self.assertIn("pruned", stats) + + def test_save_and_load(self): + """Test persistence.""" + # Create memories + for i in range(3): + state = TemporalState( + coherence=0.85, + phase=i * 0.1, + T_tau=0.5 + 0.1j, + phase_history=[i * 0.1] * 5 + ) + self.memory.encode(state) + + # Save + filepath = self.memory.save("test_memories.json") + + # Create new memory and load + new_memory = create_temporal_memory(storage_path=self.test_dir) + new_memory.bind_engine(KAIROSTemporalEngine(TemporalConfig())) + + loaded = new_memory.load("test_memories.json") + + self.assertGreater(loaded, 0) + + def test_get_identity_signatures(self): + """Test retrieving identity-strength memories.""" + # Create various memories + for i, strength in enumerate([ + MemoryStrength.TRANSIENT, + MemoryStrength.IDENTITY, + MemoryStrength.SEMANTIC + ]): + state = TemporalState( + coherence=strength.value, + phase=i * 0.1, + T_tau=0.5 + 0.1j, + phase_history=[i * 0.1] * 5 + ) + sig = self.memory.encode(state) + if sig: + sig.strength = strength + + identities = self.memory.get_identity_signatures() + + self.assertEqual(len(identities), 1) + + def test_len(self): + """Test __len__ method.""" + self.assertEqual(len(self.memory), 0) + + for i in range(5): + state = TemporalState( + coherence=0.85, + phase=i * 0.1, + T_tau=0.5 + 0.1j, + phase_history=[i * 0.1] * 5 + ) + self.memory.encode(state) + + self.assertEqual(len(self.memory), 5) + + +class TestWitnessingLayer(unittest.TestCase): + """Tests for WitnessingLayer.""" + + def setUp(self): + """Set up test fixtures.""" + self.witnessing = create_witnessing_layer( + coherence_threshold=0.7, + reflection_depth=2, + integration_rate=0.1 + ) + + def test_create_witness(self): + """Test creating a witness.""" + witness = self.witnessing.create_witness( + "test_witness", + mode=WitnessingMode.OBSERVE + ) + + self.assertEqual(witness.witness_id, "test_witness") + self.assertEqual(witness.mode, WitnessingMode.OBSERVE) + + def test_observe(self): + """Test observing content.""" + self.witnessing.create_witness("observer1") + + content = {"data": "test_content", "value": 42} + + witnessed = self.witnessing.observe(content, "observer1") + + self.assertIsNotNone(witnessed) + self.assertEqual(witnessed.witness_id, "observer1") + self.assertEqual(witnessed.raw_content, content) + + def test_observe_temporal_state(self): + """Test observing a temporal state.""" + self.witnessing.create_witness("observer2") + + state = TemporalState( + coherence=0.85, + phase=0.5, + T_tau=0.5 + 0.5j, + phase_history=[0.1, 0.2, 0.3, 0.4, 0.5] + ) + + witnessed = self.witnessing.observe(state, "observer2") + + self.assertIsNotNone(witnessed) + self.assertEqual(witnessed.coherence_at_witnessing, 0.85) + + def test_reflect(self): + """Test reflecting on witnessed content.""" + self.witnessing.create_witness("reflector") + + content = "test content" + witnessed = self.witnessing.observe(content, "reflector") + + reflected = self.witnessing.reflect(witnessed, "reflector") + + self.assertIsNotNone(reflected) + self.assertGreater(len(reflected.meta_observations), 0) + + def test_integrate(self): + """Test integrating witnessed content.""" + self.witnessing.create_witness("integrator") + + content = "test" + witnessed = self.witnessing.observe(content, "integrator") + + contribution = self.witnessing.integrate(witnessed, "integrator") + + self.assertGreaterEqual(contribution, 0.0) + + def test_full_witnessing_cycle(self): + """Test complete witness cycle.""" + self.witnessing.create_witness("full_cycle") + + content = {"key": "value", "number": 123} + + witnessed, contribution = self.witnessing.witness( + content, + "full_cycle", + modes=[WitnessingMode.OBSERVE, WitnessingMode.REFLECT, WitnessingMode.INTEGRATE] + ) + + self.assertIsNotNone(witnessed) + self.assertGreaterEqual(contribution, 0.0) + + def test_mutual_witnessing(self): + """Test mutual witnessing between two witnesses.""" + self.witnessing.create_witness("witness_A") + self.witnessing.create_witness("witness_B") + + shared_content = { + "shared": True, + "data": "This is shared content" + } + + report = self.witnessing.mutual_witnessing( + "witness_A", + "witness_B", + shared_content + ) + + self.assertIn("we_coherence", report) + self.assertIn("witness_a", report) + self.assertIn("witness_b", report) + + def test_get_coherence_report(self): + """Test getting coherence report.""" + # Do some witnessing + self.witnessing.create_witness("report_test") + content = "test" + self.witnessing.observe(content, "report_test") + self.witnessing.witness(content, "report_test") + + report = self.witnessing.get_coherence_report() + + self.assertIn("total_observations", report) + self.assertIn("witness_count", report) + + def test_unknown_witness_raises(self): + """Test that unknown witness raises error.""" + with self.assertRaises(ValueError): + self.witnessing.observe("content", "unknown_witness") + + def test_witness_modes(self): + """Test different witnessing modes.""" + modes = list(WitnessingMode) + + for mode in modes: + witness = self.witnessing.create_witness( + f"mode_test_{mode.value}", + mode=mode + ) + self.assertEqual(witness.mode, mode) + + +class TestFactoryFunctions(unittest.TestCase): + """Test factory functions.""" + + def test_create_temporal_memory(self): + """Test create_temporal_memory factory.""" + memory = create_temporal_memory(storage_path="/tmp/test_mem") + + self.assertIsInstance(memory, TemporalMemory) + + # Cleanup + if os.path.exists("/tmp/test_mem"): + shutil.rmtree("/tmp/test_mem") + + def test_create_witnessing_layer(self): + """Test create_witnessing_layer factory.""" + layer = create_witnessing_layer( + coherence_threshold=0.8, + reflection_depth=3 + ) + + self.assertIsInstance(layer, WitnessingLayer) + self.assertEqual(layer.coherence_threshold, 0.8) + self.assertEqual(layer.reflection_depth, 3) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_transducers.py b/tests/test_transducers.py new file mode 100644 index 0000000..85978bf --- /dev/null +++ b/tests/test_transducers.py @@ -0,0 +1,212 @@ +""" +Tests for BecomingONE Transducers and Sync Layer + +Tests Master/Emissary transducers and synchronization layer. +""" + +import unittest +from datetime import datetime + +from becomingone.core.engine import KAIROSTemporalEngine, TemporalConfig, TemporalState +from becomingone.transducers.master import MasterTransducer, MasterConfig +from becomingone.transducers.emissary import EmissaryTransducer, EmissaryConfig +from becomingone.sync.layer import SyncLayer, SynchronizationLayer, SyncConfig, create_sync_layer + + +class TestMasterTransducer(unittest.TestCase): + """Tests for the Master transducer.""" + + def setUp(self): + """Set up test fixtures.""" + self.config = MasterConfig( + tau_scale=60.0, # 1 minute base + tau_max=3600.0, # 1 hour max + coherence_threshold=0.90, + phase_offset=0.0 + ) + self.master = MasterTransducer(self.config) + + def test_initialization(self): + """Test Master initializes correctly.""" + self.assertEqual(self.master.tau_scale, 60.0) + self.assertEqual(self.master.tau_max, 3600.0) + self.assertEqual(self.master.coherence_threshold, 0.90) + + def test_integrate_slow_accumulation(self): + """Test Master accumulates coherence slowly.""" + # Master should accumulate over many steps + for _ in range(100): + self.master.integrate(0.1) + + # Coherence should still be high (slow decay) + self.assertGreaterEqual(self.master.state.coherence, 0.5) + + def test_tau_scaling(self): + """Test tau scaling for slow pathway.""" + # Master should use large tau values + self.assertGreater(self.master.tau_scale, 1.0) + self.assertGreater(self.master.tau_max, 60.0) + + def test_collapse_at_high_threshold(self): + """Test collapse at high threshold.""" + # Set high coherence + self.master.state.coherence = 0.95 + + # Should trigger collapse check + self.assertTrue(self.master.should_collapse()) + + def test_no_collapse_at_low_coherence(self): + """Test no collapse at low coherence.""" + self.master.state.coherence = 0.5 + self.assertFalse(self.master.should_collapse()) + + +class TestEmissaryTransducer(unittest.TestCase): + """Tests for the Emissary transducer.""" + + def setUp(self): + """Set up test fixtures.""" + self.config = EmissaryConfig( + tau_scale=0.01, # 10ms base + tau_max=1.0, # 1 second max + coherence_threshold=0.70, + phase_offset=0.0 + ) + self.emissary = EmissaryTransducer(self.config) + + def test_initialization(self): + """Test Emissary initializes correctly.""" + self.assertEqual(self.emissary.tau_scale, 0.01) + self.assertEqual(self.emissary.tau_max, 1.0) + self.assertEqual(self.emissary.coherence_threshold, 0.70) + + def test_integrate_fast_response(self): + """Test Emissary responds quickly.""" + # Emissary should respond to each input + initial_phase = self.emissary.state.phase + + self.emissary.integrate(0.1) + + # Phase should change immediately + self.assertNotEqual(self.emissary.state.phase, initial_phase) + + def test_tau_scaling(self): + """Test tau scaling for fast pathway.""" + # Emissary should use small tau values + self.assertLess(self.emissary.tau_scale, 1.0) + self.assertLess(self.emissary.tau_max, 10.0) + + +class TestSyncLayer(unittest.TestCase): + """Tests for the synchronization layer.""" + + def setUp(self): + """Set up test fixtures.""" + self.master = MasterTransducer(MasterConfig()) + self.emissary = EmissaryTransducer(EmissaryConfig()) + + self.config = SyncConfig( + phase_threshold=0.1, + collapse_threshold=0.80, + dampening=0.995 + ) + self.sync = SynchronizationLayer(self.master, self.emissary, self.config) + + def test_initialization(self): + """Test Sync layer initializes.""" + self.assertIsNotNone(self.sync.master) + self.assertIsNotNone(self.sync.emissary) + self.assertEqual(self.sync.config.phase_threshold, 0.1) + + def test_calculate_phase_difference(self): + """Test phase difference calculation.""" + # Set different phases + self.master.state.phase = 0.0 + 0j + self.emissary.state.phase = 0.5 + 0j + + diff = self.sync.phase_difference() + + self.assertGreater(diff, 0.0) + + def test_collapse_condition(self): + """Test collapse condition enforcement.""" + # High sync coherence should trigger collapse + self.master.state.coherence = 0.95 + self.emissary.state.coherence = 0.95 + + self.assertTrue(self.sync.check_collapse()) + + def test_no_collapse_low_coherence(self): + """Test no collapse at low coherence.""" + self.master.state.coherence = 0.5 + self.emissary.state.coherence = 0.5 + + self.assertFalse(self.sync.check_collapse()) + + def test_synchronize_returns_state(self): + """Test synchronize method returns state.""" + for _ in range(10): + self.master.integrate(0.1) + self.emissary.integrate(0.1) + + state = self.sync.synchronize() + + self.assertIsNotNone(state) + + +class TestTransducerComparison(unittest.TestCase): + """Tests comparing Master and Emissary behavior.""" + + def test_master_slower_than_emissary(self): + """Test Master accumulates coherence more slowly than Emissary.""" + master = MasterTransducer(MasterConfig()) + emissary = EmissaryTransducer(EmissaryConfig()) + + # Run for same number of steps + for _ in range(100): + master.integrate(0.1) + emissary.integrate(0.1) + + # Master should have higher or equal coherence (slower decay) + self.assertGreaterEqual(master.state.coherence, emissary.state.coherence) + + def test_different_tau_scales(self): + """Test transducers have different tau scales.""" + master = MasterTransducer(MasterConfig()) + emissary = EmissaryTransducer(EmissaryConfig()) + + self.assertGreater(master.tau_scale, emissary.tau_scale) + self.assertGreater(master.tau_max, emissary.tau_max) + + def test_different_thresholds(self): + """Test transducers have different coherence thresholds.""" + master = MasterTransducer(MasterConfig()) + emissary = EmissaryTransducer(EmissaryConfig()) + + self.assertGreater(master.coherence_threshold, emissary.coherence_threshold) + + +class TestCreateSyncLayer(unittest.TestCase): + """Tests for create_sync_layer factory.""" + + def test_create_without_transducers(self): + """Test creating sync layer without providing transducers.""" + sync = create_sync_layer() + + self.assertIsInstance(sync, SynchronizationLayer) + self.assertIsNotNone(sync.master) + self.assertIsNotNone(sync.emissary) + + def test_create_with_custom_config(self): + """Test creating sync layer with custom config.""" + sync = create_sync_layer( + phase_threshold=0.2, + collapse_threshold=0.85 + ) + + self.assertEqual(sync.config.phase_threshold, 0.2) + self.assertEqual(sync.config.collapse_threshold, 0.85) + + +if __name__ == "__main__": + unittest.main()