"""Pure synthetic records, deliberately not a neural simulation or file test.""" import hashlib import json import unittest import numpy as np import engineering_scopes as s def sha(array): return hashlib.sha256(np.ascontiguousarray(array).tobytes()).hexdigest() def record(start, stop, *, reference=True, blank_requests=False): n, k, length = 3, 2, stop - start ticks = np.array([0, 7, 19, 21, 41, 61], np.int32) nodes = np.array([0, 1, 2, 0, 2, 1], np.int32) selected = (ticks >= start) & (ticks < stop) event_ticks, event_nodes = ticks[selected] - np.int32(start), nodes[selected].copy() v = np.array([-.05, -.04, -.03], np.float64) g = np.array([-0., float(stop - 1), -.001], np.float64) ref = np.zeros(n, np.int32) state = dict(step=event_ticks, neuron_idx=event_nodes, counts=np.bincount(event_nodes, minlength=n).astype(np.int32), v=v, g=g, ref=ref) raw_global = np.zeros((19, n), np.uint8) for tick, node in zip(ticks, nodes): if max(0, stop - 19) <= tick < stop: raw_global[tick % 19, node] = 1 normalized = raw_global[(np.arange(19) + stop % 19) % 19] raw_local = normalized[(np.arange(19) - length % 19) % 19].copy() generated_all = np.repeat(np.arange(1, stop + 1, dtype=np.float32)[:, None], k, axis=1) applied_all = np.concatenate((np.zeros((18, k), np.float32), generated_all))[:stop] pending = np.zeros((18, k), np.float32) kept = min(stop, 18); pending[-kept:] = generated_all[-kept:] apl_state = np.full(2, np.float32(stop + 1), np.float32) terminal = dict(spike_ring=raw_local, apl_state=apl_state, apl_pending=pending, next_tick=np.asarray(length, np.int64)) trace = dict(generated=generated_all[start:stop].copy(), applied=applied_all[start:stop].copy(), drive=np.repeat(np.arange(start, stop, dtype=np.float32)[:, None], 2, axis=1), state=np.repeat(np.arange(start + 2, stop + 2, dtype=np.float32)[:, None], 2, axis=1)) hashes, queues = [], [] for local, global_tick in enumerate(range(start, stop)): hashes.append(dict(tick=local, v_sha256=sha(v), g_sha256=sha(np.array([-0., float(global_tick), -.001], np.float64)), ref_sha256=sha(ref))) edges, accepted = np.array([global_tick], np.int64), np.array([global_tick % 2], np.uint8) queues.append(dict(tick=local, count=1, accepted_count=int(accepted.sum()), ordered_edge_sha256=sha(edges), accepted_mask_sha256=sha(accepted), joint_sha256=hashlib.sha256( str(local).encode() + edges.tobytes() + accepted.tobytes()).hexdigest())) requests = np.array([[int(t), int(i), 1, 1, 0] for t, i in zip(event_ticks, event_nodes)], np.int64).reshape(-1, 5) if blank_requests: requests = np.empty((0, 5), np.int64) summary = dict(complete=True, pass_all=True, all_finite=True, steps=length, completed_ticks=length, nodes=n, precision='float64', apl_precision='float32', apl_delay_steps=18, checkpoint_continuation_used=start > 0, global_tick_offset=start, global_next_tick=stop, effective_weights_raw_sha256='e' * 64, diagnostic_mode='reference' if reference else 'light', full_APL_history_recorded=reference, full_tick_state_hashes_recorded=reference, full_queue_hashes_recorded=reference) return dict(source='synthetic', state=state, terminal=terminal, requests=requests, steps=length, global_tick_offset=start, summary=summary, traces=trace if reference else {}, state_hashes=hashes if reference else None, queues=queues if reference else None, terminal_origin='synthetic_recorded') class SliceTests(unittest.TestCase): def test_reference_slice_all_observations_exact_and_inputs_unchanged(self): candidate, whole = record(20, 44), record(0, 64) before = candidate['terminal']['spike_ring'].tobytes(), whole['requests'].tobytes() report = s.compare_slice(candidate, whole) self.assertTrue(report['numerical_exact'], report['failed_checks']) self.assertTrue(report['observed_scope']['candidate_per_tick_state_hashes']) self.assertTrue(report['observed_scope']['candidate_per_tick_queue_fields']) self.assertFalse(report['observed_scope']['queue_joint_hashes_compared']) self.assertNotEqual(candidate['queues'][0]['joint_sha256'], whole['queues'][20]['joint_sha256']) self.assertEqual(before, (candidate['terminal']['spike_ring'].tobytes(), whole['requests'].tobytes())) json.dumps(report, allow_nan=False) def test_light_uses_final_raw_hash_but_does_not_fabricate_optional_history(self): report = s.compare_slice(record(20, 44, reference=False), record(0, 64)) self.assertTrue(report['numerical_exact']) self.assertTrue(report['observed_scope']['final_state_against_reference_tick_hash']) for name in ('candidate_APL_histories', 'candidate_per_tick_state_hashes', 'candidate_per_tick_queue_fields'): self.assertFalse(report['observed_scope'][name]) self.assertNotIn('slice.per_tick_state_hashes', report['checks']) self.assertTrue(any('unobserved' in item for item in report['limitations'])) def test_short_fresh_positive_zero_padding_and_prefix_events(self): self.assertTrue(s.compare_slice(record(0, 5), record(0, 64))['numerical_exact']) candidate = record(0, 5, reference=False) candidate['terminal']['apl_pending'][0, 0] = np.float32(-0.) report = s.compare_slice(candidate, record(0, 64)) self.assertFalse(report['checks']['slice.APL_pending']) def test_blank_requests_are_valid_but_wrong_acceptance_is_not_equal(self): self.assertTrue(s.compare_slice(record(20, 44, blank_requests=True), record(0, 64, blank_requests=True))['numerical_exact']) candidate = record(20, 44) candidate['requests'][0, 3:] = [0, 1] report = s.compare_slice(candidate, record(0, 64)) self.assertFalse(report['checks']['slice.requests_five_fields']) def test_wrong_count_or_final_hash_binding_is_invalid_not_a_pass(self): for mutation in ('count', 'terminal', 'last_hash'): candidate = record(20, 44) if mutation == 'count': candidate['state']['counts'][0] += 1 elif mutation == 'terminal': candidate['state']['g'][1] += 1. else: candidate['state_hashes'][-1]['g_sha256'] = '0' * 64 with self.subTest(mutation=mutation), self.assertRaises(ValueError): s.compare_slice(candidate, record(0, 64)) def test_signed_zero_light_terminal_vs_reference_hash_is_detected(self): candidate = record(20, 44, reference=False) candidate['state']['g'][0] = 0. report = s.compare_slice(candidate, record(0, 64)) self.assertFalse(report['checks']['slice.final_g_raw_hash']) def test_interior_trace_state_hash_and_queue_mismatch_are_not_hidden(self): candidate = record(20, 44) candidate['traces']['drive'][1, 0] += np.float32(1.) candidate['state_hashes'][1]['v_sha256'] = '0' * 64 candidate['queues'][1]['accepted_mask_sha256'] = '1' * 64 report = s.compare_slice(candidate, record(0, 64)) for key in ('slice.APL_trace.drive', 'slice.per_tick_state_hashes', 'slice.queue_order_and_acceptance'): self.assertFalse(report['checks'][key]) def test_missing_reference_observations_and_invalid_bounds_fail_closed(self): for mutation in ('light_mono', 'light_labeled_full_mono', 'mono_hashes', 'candidate_traces', 'partial_trace', 'out_of_bounds', 'nonzero_mono'): candidate, mono = record(20, 44), record(0, 64) if mutation == 'light_mono': mono = record(0, 64, reference=False) elif mutation == 'light_labeled_full_mono': mono['summary']['diagnostic_mode'] = 'light' elif mutation == 'mono_hashes': mono['state_hashes'] = None elif mutation == 'candidate_traces': candidate['traces'] = {} elif mutation == 'partial_trace': candidate['traces'].pop('drive') elif mutation == 'out_of_bounds': candidate = record(60, 65) else: mono = record(1, 65) with self.subTest(mutation=mutation), self.assertRaises(ValueError): s.compare_slice(candidate, mono) def test_summary_type_scope_nonfinite_and_paths_rejected(self): for mutation in ('bool', 'failed', 'offset', 'mode', 'weight', 'nan'): candidate = record(20, 44) if mutation == 'bool': candidate['summary']['steps'] = True elif mutation == 'failed': candidate['summary']['pass_all'] = False elif mutation == 'offset': candidate['summary']['global_tick_offset'] = 0 elif mutation == 'mode': candidate['summary']['diagnostic_mode'] = 'unknown' elif mutation == 'weight': candidate['summary'].pop('effective_weights_raw_sha256') else: candidate['state']['g'][0] = np.nan with self.subTest(mutation=mutation), self.assertRaises(ValueError): s.compare_slice(candidate, record(0, 64)) with self.assertRaises(ValueError): s.compare_slice('/never/open/this', record(0, 64)) class CheckpointTests(unittest.TestCase): def test_local_phase_normalization_without_segment_cumulative_counts_confusion(self): candidate, prefix = record(20, 64), record(0, 64) self.assertFalse(s.ec.bit_equal(candidate['terminal']['spike_ring'], prefix['terminal']['spike_ring'])) self.assertFalse(s.ec.bit_equal(candidate['state']['counts'], prefix['state']['counts'])) report = s.compare_checkpoint(candidate, prefix) self.assertTrue(report['numerical_exact'], report['failed_checks']) self.assertFalse(report['observed_scope']['counts_compared']) self.assertFalse(report['observed_scope']['per_tick_state_histories_compared']) def test_signed_zero_different_stop_or_effective_weights_cannot_pass(self): for mutation in ('zero', 'stop', 'weight'): candidate, prefix = record(20, 64, reference=False), record(0, 64, reference=False) if mutation == 'zero': candidate['state']['g'][0] = 0. elif mutation == 'stop': prefix = record(0, 63, reference=False) else: candidate['summary']['effective_weights_raw_sha256'] = 'f' * 64 with self.subTest(mutation=mutation): self.assertFalse(s.compare_checkpoint(candidate, prefix)['numerical_exact']) class ParentCarryTests(unittest.TestCase): def test_short_parent_carry_observed_and_json_safe(self): report = s.compare_parent_carry(record(20, 25), record(0, 20)) self.assertTrue(report['numerical_exact'], report['failed_checks']) self.assertEqual(report['observed_scope']['unwritten_spike_slot_count'], 14) self.assertEqual(report['observed_scope']['unconsumed_APL_row_count'], 13) self.assertEqual(report['observed_scope']['applied_parent_APL_rows_compared'], 5) json.dumps(report, allow_nan=False) def test_unwritten_or_unconsumed_corruption_detected_even_if_terminal_is_self_consistent(self): for mutation in ('ring', 'pending', 'applied'): candidate = record(20, 25) if mutation == 'ring': candidate['terminal']['spike_ring'][10, 0] ^= np.uint8(1) elif mutation == 'pending': candidate['terminal']['apl_pending'][0, 0] += np.float32(1.) else: candidate['traces']['applied'][0, 0] += np.float32(1.) with self.subTest(mutation=mutation): self.assertFalse(s.compare_parent_carry(candidate, record(0, 20))['numerical_exact']) def test_light_missing_optional_applied_trace_is_not_fabricated(self): report = s.compare_parent_carry(record(20, 25, reference=False), record(0, 20)) self.assertTrue(report['numerical_exact']) self.assertFalse(report['observed_scope']['applied_parent_APL_prefix_checked']) self.assertIsNone(report['observed_scope']['applied_parent_APL_rows_compared']) self.assertNotIn('parent.applied_APL_prefix', report['checks']) def test_long_segment_discloses_no_unwritten_slots_and_wrong_parent_clock_fails(self): report = s.compare_parent_carry(record(20, 44), record(0, 20)) self.assertTrue(report['numerical_exact']) self.assertFalse(report['observed_scope']['unwritten_spike_slots_checked']) self.assertFalse(report['observed_scope']['unconsumed_APL_prefix_checked']) self.assertEqual(report['observed_scope']['applied_parent_APL_rows_compared'], 18) self.assertFalse(s.compare_parent_carry(record(20, 44), record(0, 19))['numerical_exact']) with self.assertRaises(ValueError): s.compare_parent_carry(record(0, 5), record(0, 20)) def test_long_light_carry_scope_explicitly_has_no_observed_carry_array(self): report = s.compare_parent_carry(record(20, 44, reference=False), record(0, 20)) self.assertTrue(report['numerical_exact']) self.assertFalse(report['observed_scope']['any_carry_values_directly_compared']) self.assertTrue(any('clock/shape/weight identity only' in line for line in report['limitations'])) self.assertFalse(report['complete_dynamical_equivalence_proven']) if __name__ == '__main__': unittest.main()