#!/usr/bin/env python3 # UNDF: UNDF-2026-000001259 (lean4-0001), UNDF-2026-000001260 (lean4-0002), # UNDF-2026-000001261 (lean4-0003), UNDF-2026-000001262 (lean4-0004), # UNDF-2026-000001263 (lean4-0005), UNDF-2026-000001264 (lean4-0006), # UNDF-2026-000001265 (lean4-0007) # # CWE-407: Algorithmic Complexity (lean4-0001, 0002, 0003, 0007) # CWE-362: Race Condition (lean4-0004, 0005) # CWE-668: Leaked Context (lean4-0006) # # Defects: # lean4-0001: List.contains O(N) inside O(N) guardCycle loop -- O(N^2) total. # lean4-0002: std::find(result_args) O(N) inside O(K) to_check loop -- O(K*N) total. # Also std::find(m_lparams) O(L) per while iteration -- O(L^2) total. # lean4-0003: std::find(lp_names) O(N) per while iteration -- O(N^2) total. # lean4-0004: double-checked locking race: shared_lock read, unlock, unique_lock upgrade # leaves a window where g_native_symbol_cache can be modified mid-read. # lean4-0005: IO.Ref.modify not atomic across concurrent tasks -- concurrent pushes # lose job registrations. # lean4-0006: LEAN_THREAD_PTR(g_opts) not reset at task boundaries -- thread-pool # reuse leaks prior elaboration task's trace options into next task. # lean4-0007: new_env_vars.count({key_begin, key_end}) constructs std::string from # char* range on every O(N) iteration -- O(N^2) total over N env vars. # # Fixes: # lean4-0001: CycleT carries (HashSet x List). guardCycle calls stackContains O(1). # lean4-0002: expr_set built once from result_args before loop -- O(1) lookup. # name_set built once from m_lparams before while loop -- O(1) lookup. # lean4-0003: lp_set built once from lp_names before while loop -- O(1) lookup. # lean4-0004: single unique_lock acquisition; eliminates shared->unique upgrade race. # lean4-0005: IO.Mutex (Array OpaqueJob) with atomically for all register + poll ops. # lean4-0006: register_thread_local_reset_fn sets g_opts = nullptr before each task. # lean4-0007: std::unordered_set override_keys built once -- O(1) lookup. # # Complexity gates (from bench/results.txt on this machine): # lean4-0001: N=2000 defective=38ms, fixed=1.1ms. Fixed must complete in <20ms. # k-scaling: time(2N) / time(N) must be <3x (O(N) not O(N^2)). # lean4-0002: K=N=1000 defective=14ms, fixed=0.021ms. Fixed must complete in <5ms. # k-scaling: time(2K) / time(K) must be <3x (O(K) not O(K^2)). # lean4-0003: N=1000 defective=19.9ms, fixed=0.095ms. Fixed must complete in <5ms. # k-scaling: time(2N) / time(N) must be <3x (O(N) not O(N^2)). # lean4-0007: N=500 env vars. Fixed must complete in <1ms. import sys import time import threading import unittest PASS = "PASS" FAIL = "FAIL" _results = [] def record(name, passed, detail=""): tag = PASS if passed else FAIL msg = f" [{tag}] {name}" if detail: msg += f" -- {detail}" print(msg) _results.append(passed) # --------------------------------------------------------------------------- # lean4-0001: guardCycle List.contains vs HashSet # --------------------------------------------------------------------------- def _guardcycle_defective(n): """ O(N^2): list membership inside O(N) traversal -- models List.contains. Uses list.append (not insert) to isolate membership cost from list mutation cost. The defect is the O(N) membership check, not the stack push. """ parents = [] for i in range(n): _ = i in parents # O(len(parents)) -- this is the defect parents.append(i) # O(1) append keeps mutation cost constant def _guardcycle_fixed(n): """O(N): set membership inside O(N) traversal -- models HashSet.contains.""" parents_set = set() for i in range(n): _ = i in parents_set # O(1) parents_set.add(i) class TestLean40001Unit(unittest.TestCase): """lean4-0001 Unit: list membership vs set membership -- same answer, small scale.""" def test_no_cycle_detected_list(self): parents = [1, 2, 3] result = 4 in parents record("0001-unit: key not in list returns False", result is False) self.assertFalse(result) def test_cycle_detected_list(self): parents = [1, 2, 3] result = 2 in parents record("0001-unit: key in list returns True", result is True) self.assertTrue(result) def test_no_cycle_detected_set(self): parents_set = {1, 2, 3} result = 4 in parents_set record("0001-unit: key not in set returns False", result is False) self.assertFalse(result) def test_cycle_detected_set(self): parents_set = {1, 2, 3} result = 2 in parents_set record("0001-unit: key in set returns True", result is True) self.assertTrue(result) def test_list_and_set_agree_small(self): """Both representations agree on membership for every key in a small traversal.""" n = 20 parents_list = [] parents_set = set() ok = True for i in range(n): list_answer = i in parents_list set_answer = i in parents_set if list_answer != set_answer: ok = False break parents_list.insert(0, i) parents_set.add(i) record("0001-unit: list and set agree on all membership queries (n=20)", ok) self.assertTrue(ok) def test_cycle_detected_same_result(self): """Simulated guardCycle with duplicate key produces cycle via both approaches.""" keys = [1, 2, 3, 2] # key 2 appears twice -- cycle # list approach parents_list = [] cycle_list = None for k in keys: if k in parents_list: cycle_list = k break parents_list.insert(0, k) # set approach parents_set = set() cycle_set = None for k in keys: if k in parents_set: cycle_set = k break parents_set.add(k) ok = cycle_list == cycle_set == 2 record("0001-unit: cycle key detected identically by list and set", ok, f"list={cycle_list} set={cycle_set}") self.assertEqual(cycle_list, cycle_set) class TestLean40001Integration(unittest.TestCase): """lean4-0001 Integration: full traversal correctness at medium scale.""" def test_linear_chain_no_cycle_list(self): n = 200 parents = [] found_cycle = False for i in range(n): if i in parents: found_cycle = True break parents.insert(0, i) record("0001-intg: linear chain n=200 no false cycle (list)", not found_cycle) self.assertFalse(found_cycle) def test_linear_chain_no_cycle_set(self): n = 200 parents_set = set() parents_list = [] found_cycle = False for i in range(n): if i in parents_set: found_cycle = True break parents_set.add(i) parents_list.insert(0, i) record("0001-intg: linear chain n=200 no false cycle (set)", not found_cycle) self.assertFalse(found_cycle) def test_list_and_set_agree_medium(self): n = 500 parents_list = [] parents_set = set() mismatches = 0 for i in range(n): if (i in parents_list) != (i in parents_set): mismatches += 1 parents_list.insert(0, i) parents_set.add(i) ok = mismatches == 0 record(f"0001-intg: list/set agree on all queries n={n}", ok, f"mismatches={mismatches}") self.assertEqual(mismatches, 0) def test_cycle_at_midpoint_list_vs_set(self): """Insert duplicate at midpoint and confirm both approaches detect at same position.""" n = 100 keys = list(range(n)) + [50] # duplicate of key 50 cycle_pos_list = None parents_list = [] for pos, k in enumerate(keys): if k in parents_list: cycle_pos_list = pos break parents_list.insert(0, k) cycle_pos_set = None parents_set = set() for pos, k in enumerate(keys): if k in parents_set: cycle_pos_set = pos break parents_set.add(k) ok = cycle_pos_list == cycle_pos_set == n record("0001-intg: cycle at midpoint detected at same position", ok, f"list_pos={cycle_pos_list} set_pos={cycle_pos_set}") self.assertEqual(cycle_pos_list, cycle_pos_set) def test_call_stack_ordering_preserved(self): """Set membership does not affect list ordering for error reporting.""" n = 50 parents_list = [] parents_set = set() for i in range(n): parents_list.insert(0, i) parents_set.add(i) # list remains reverse-insertion order; set membership independent of order ok = parents_list[0] == n - 1 and parents_list[-1] == 0 record("0001-intg: call stack list preserves insertion order", ok) self.assertTrue(ok) class TestLean40001FunctionalComplexityGate(unittest.TestCase): """lean4-0001 Complexity gate: O(N) fixed path must complete well under O(N^2) threshold.""" def _time_defective(self, n, trials=3): times = [] for _ in range(trials): t0 = time.perf_counter() _guardcycle_defective(n) times.append(time.perf_counter() - t0) return min(times) def _time_fixed(self, n, trials=3): times = [] for _ in range(trials): t0 = time.perf_counter() _guardcycle_fixed(n) times.append(time.perf_counter() - t0) return min(times) def test_fixed_n2000_under_20ms(self): """Fixed path N=2000 must complete in <20ms (benchmark: 1.1ms).""" t = self._time_fixed(2000) * 1000 ok = t < 20.0 record(f"0001-gate: fixed N=2000 <20ms", ok, f"{t:.3f}ms") self.assertLess(t, 20.0, f"fixed N=2000 took {t:.3f}ms, expected <20ms") def test_fixed_scaling_linear(self): """Fixed path: time(N=2000) / time(N=1000) must be <3x (O(N) not O(N^2)).""" t1 = self._time_fixed(1000) t2 = self._time_fixed(2000) ratio = t2 / t1 if t1 > 0 else 0 ok = ratio < 3.0 record(f"0001-gate: fixed k-scaling <3x (got {ratio:.2f}x)", ok, f"t(1000)={t1*1000:.3f}ms t(2000)={t2*1000:.3f}ms") self.assertLess(ratio, 3.0, f"fixed time ratio {ratio:.2f}x >= 3x -- O(N^2) regression detected") def test_defective_scaling_quadratic(self): """Defective path: time(N=1000) / time(N=500) must be >3x (confirms O(N^2) model).""" t1 = self._time_defective(500) t2 = self._time_defective(1000) ratio = t2 / t1 if t1 > 0 else 0 ok = ratio > 3.0 record(f"0001-gate: defective scaling >3x confirms O(N^2) (got {ratio:.2f}x)", ok, f"t(500)={t1*1000:.3f}ms t(1000)={t2*1000:.3f}ms") self.assertGreater(ratio, 3.0, f"defective ratio {ratio:.2f}x not >3x -- benchmark model may be wrong") def test_fixed_faster_than_defective(self): """Fixed path N=2000 must be faster than defective N=2000.""" td = self._time_defective(2000) * 1000 tf = self._time_fixed(2000) * 1000 ok = tf < td record(f"0001-gate: fixed faster than defective N=2000", ok, f"defective={td:.3f}ms fixed={tf:.3f}ms speedup={td/tf:.1f}x") self.assertLess(tf, td) # --------------------------------------------------------------------------- # lean4-0002: expr_set inductive type check std::find vs set # --------------------------------------------------------------------------- def _inductive_defective(k, n): """O(K*N): list.contains inside O(K) loop -- models std::find(result_args).""" to_check = list(range(k)) result_args = list(range(n, 2 * n)) # disjoint: scan always goes full length for arg in to_check: _ = arg in result_args # O(N) def _inductive_fixed(k, n): """O(K+N): set built once, O(1) lookup per item.""" to_check = list(range(k)) result_args = list(range(n, 2 * n)) result_set = set(result_args) # O(N) once for arg in to_check: _ = arg in result_set # O(1) class TestLean40002Unit(unittest.TestCase): """lean4-0002 Unit: std::find vs expr_set membership -- same answer, small scale.""" def test_find_present_list(self): result_args = [10, 20, 30] ok = 20 in result_args record("0002-unit: target present in list", ok) self.assertTrue(ok) def test_find_absent_list(self): result_args = [10, 20, 30] ok = 99 not in result_args record("0002-unit: target absent from list", ok) self.assertTrue(ok) def test_find_present_set(self): result_set = {10, 20, 30} ok = 20 in result_set record("0002-unit: target present in set", ok) self.assertTrue(ok) def test_find_absent_set(self): result_set = {10, 20, 30} ok = 99 not in result_set record("0002-unit: target absent from set", ok) self.assertTrue(ok) def test_list_and_set_agree_small(self): """List and set give identical membership answers for every probe.""" result_args = list(range(20)) result_set = set(result_args) probes = list(range(30)) mismatches = sum(1 for p in probes if (p in result_args) != (p in result_set)) ok = mismatches == 0 record("0002-unit: list and set agree on all probes (20 items, 30 probes)", ok, f"mismatches={mismatches}") self.assertEqual(mismatches, 0) def test_condition2_both_return_same_fail(self): """Condition 2 check: arg not in result_args returns same result for list vs set.""" result_args = [1, 2, 3, 4, 5] result_set = set(result_args) to_check = [1, 3, 99] # 99 not present -- condition 2 fails list_fail = next((a for a in to_check if a not in result_args), None) set_fail = next((a for a in to_check if a not in result_set), None) ok = list_fail == set_fail == 99 record("0002-unit: condition2 fail detected identically by list and set", ok, f"list={list_fail} set={set_fail}") self.assertEqual(list_fail, set_fail) class TestLean40002Integration(unittest.TestCase): """lean4-0002 Integration: full to_check sweep correctness at medium scale.""" def test_all_present_list(self): n = 300 result_args = list(range(n)) to_check = list(range(n)) failures = [a for a in to_check if a not in result_args] ok = len(failures) == 0 record(f"0002-intg: all K={n} args found in result_args list", ok, f"failures={len(failures)}") self.assertEqual(len(failures), 0) def test_all_present_set(self): n = 300 result_set = set(range(n)) to_check = list(range(n)) failures = [a for a in to_check if a not in result_set] ok = len(failures) == 0 record(f"0002-intg: all K={n} args found in result_set", ok, f"failures={len(failures)}") self.assertEqual(len(failures), 0) def test_list_and_set_agree_medium(self): n = 500 result_args = list(range(n)) result_set = set(result_args) to_check = list(range(n + 100)) # some hits, some misses mismatches = sum( 1 for a in to_check if (a in result_args) != (a in result_set) ) ok = mismatches == 0 record(f"0002-intg: list/set agree K={n+100} probes over N={n} items", ok, f"mismatches={mismatches}") self.assertEqual(mismatches, 0) def test_disjoint_all_fail_consistently(self): """When to_check and result_args are disjoint, both list and set return all-miss.""" n = 200 result_args = list(range(n)) result_set = set(result_args) to_check = list(range(n, 2 * n)) # fully disjoint list_hits = sum(1 for a in to_check if a in result_args) set_hits = sum(1 for a in to_check if a in result_set) ok = list_hits == set_hits == 0 record(f"0002-intg: disjoint sets all-miss consistently n={n}", ok, f"list_hits={list_hits} set_hits={set_hits}") self.assertEqual(list_hits, set_hits) def test_partial_overlap(self): """Half overlap: list and set detect same failing args.""" n = 100 result_args = list(range(n)) result_set = set(result_args) to_check = list(range(n // 2, n + n // 2)) # half in, half out list_misses = [a for a in to_check if a not in result_args] set_misses = [a for a in to_check if a not in result_set] ok = list_misses == set_misses record(f"0002-intg: partial overlap misses match list vs set n={n}", ok, f"list={len(list_misses)} set={len(set_misses)}") self.assertEqual(list_misses, set_misses) class TestLean40002FunctionalComplexityGate(unittest.TestCase): """lean4-0002 Complexity gate: O(K+N) fixed path must complete well under O(K*N).""" def _time_defective(self, k, n, trials=3): times = [] for _ in range(trials): t0 = time.perf_counter() _inductive_defective(k, n) times.append(time.perf_counter() - t0) return min(times) def _time_fixed(self, k, n, trials=3): times = [] for _ in range(trials): t0 = time.perf_counter() _inductive_fixed(k, n) times.append(time.perf_counter() - t0) return min(times) def test_fixed_kn1000_under_5ms(self): """Fixed path K=N=1000 must complete in <5ms (benchmark: 0.021ms).""" t = self._time_fixed(1000, 1000) * 1000 ok = t < 5.0 record(f"0002-gate: fixed K=N=1000 <5ms", ok, f"{t:.3f}ms") self.assertLess(t, 5.0, f"fixed K=N=1000 took {t:.3f}ms, expected <5ms") def test_fixed_scaling_linear(self): """Fixed path: time(K=N=1000) / time(K=N=500) must be <3x (O(K+N) not O(K*N)).""" t1 = self._time_fixed(500, 500) t2 = self._time_fixed(1000, 1000) ratio = t2 / t1 if t1 > 0 else 0 ok = ratio < 3.0 record(f"0002-gate: fixed k-scaling <3x (got {ratio:.2f}x)", ok, f"t(500)={t1*1000:.3f}ms t(1000)={t2*1000:.3f}ms") self.assertLess(ratio, 3.0, f"fixed time ratio {ratio:.2f}x >= 3x -- O(K*N) regression detected") def test_defective_scaling_quadratic(self): """Defective path: time(K=N=500) / time(K=N=250) must be >3x (confirms O(K*N)).""" t1 = self._time_defective(250, 250) t2 = self._time_defective(500, 500) ratio = t2 / t1 if t1 > 0 else 0 ok = ratio > 3.0 record(f"0002-gate: defective scaling >3x confirms O(K*N) (got {ratio:.2f}x)", ok, f"t(250)={t1*1000:.3f}ms t(500)={t2*1000:.3f}ms") self.assertGreater(ratio, 3.0, f"defective ratio {ratio:.2f}x not >3x -- benchmark model may be wrong") def test_fixed_faster_than_defective(self): """Fixed path K=N=1000 must be faster than defective K=N=1000.""" td = self._time_defective(1000, 1000) * 1000 tf = self._time_fixed(1000, 1000) * 1000 ok = tf < td record(f"0002-gate: fixed faster than defective K=N=1000", ok, f"defective={td:.3f}ms fixed={tf:.3f}ms speedup={td/tf:.1f}x") self.assertLess(tf, td) # --------------------------------------------------------------------------- # lean4-0003: mk_fresh_lp_name while loop # --------------------------------------------------------------------------- def _fresh_name_defective(n): """O(N^2): list.contains O(N) per while iteration, N iterations worst case. Uses string keys to avoid CPython small-int cache optimizations that mask O(N^2).""" existing = [f"l{i}" for i in range(n)] candidate = f"l{n}" # not in list -- must scan full list every probe for _ in range(n): _ = candidate in existing # O(N) each time def _fresh_name_fixed(n): """O(N): set built once O(N), O(1) per while iteration.""" existing = [f"l{i}" for i in range(n)] existing_set = set(existing) # O(N) build once candidate = f"l{n}" for _ in range(n): _ = candidate in existing_set # O(1) class TestLean40003Unit(unittest.TestCase): """lean4-0003 Unit: while-loop name collision check list vs set.""" def test_name_not_in_list(self): names = ["l1", "l2", "l3"] ok = "l4" not in names record("0003-unit: fresh candidate not in list", ok) self.assertTrue(ok) def test_name_in_list(self): names = ["l1", "l2", "l3"] ok = "l2" in names record("0003-unit: collision detected in list", ok) self.assertTrue(ok) def test_name_not_in_set(self): name_set = {"l1", "l2", "l3"} ok = "l4" not in name_set record("0003-unit: fresh candidate not in set", ok) self.assertTrue(ok) def test_name_in_set(self): name_set = {"l1", "l2", "l3"} ok = "l2" in name_set record("0003-unit: collision detected in set", ok) self.assertTrue(ok) def test_list_and_set_agree_small(self): names = [f"l{i}" for i in range(20)] name_set = set(names) probes = [f"l{i}" for i in range(25)] mismatches = sum(1 for p in probes if (p in names) != (p in name_set)) ok = mismatches == 0 record("0003-unit: list and set agree on all name probes (20 names, 25 probes)", ok, f"mismatches={mismatches}") self.assertEqual(mismatches, 0) def test_fresh_name_found_same(self): """Both approaches find the same fresh name (first not in existing).""" existing = [f"l{i}" for i in range(1, 6)] # l1..l5 taken existing_set = set(existing) # list approach: scan l1..l6 until free candidate_list = None for i in range(1, 20): n = f"l{i}" if n not in existing: candidate_list = n break # set approach candidate_set = None for i in range(1, 20): n = f"l{i}" if n not in existing_set: candidate_set = n break ok = candidate_list == candidate_set == "l6" record("0003-unit: list and set find same fresh name", ok, f"list={candidate_list} set={candidate_set}") self.assertEqual(candidate_list, candidate_set) class TestLean40003Integration(unittest.TestCase): """lean4-0003 Integration: fresh name generation correctness at medium scale.""" def test_fresh_name_not_in_existing(self): n = 200 existing = [f"l{i}" for i in range(1, n + 1)] existing_set = set(existing) # simulate: start at l1, increment until free i = 1 while f"l{i}" in existing_set: i += 1 fresh = f"l{i}" ok = fresh not in existing_set and fresh == f"l{n+1}" record(f"0003-intg: fresh name not in existing n={n}", ok, f"fresh={fresh}") self.assertTrue(ok) def test_list_and_set_same_result_medium(self): n = 300 existing = list(range(n)) existing_set = set(existing) probes = list(range(n + 50)) mismatches = sum(1 for p in probes if (p in existing) != (p in existing_set)) ok = mismatches == 0 record(f"0003-intg: list/set agree n={n} with {n+50} probes", ok, f"mismatches={mismatches}") self.assertEqual(mismatches, 0) def test_dense_collisions_same_outcome(self): """When all candidates collide until the very last slot, both find it.""" n = 100 # l1..l{n} all taken; fresh = l{n+1} existing = [f"l{i}" for i in range(1, n + 1)] existing_set = set(existing) # list approach i = 1 while f"l{i}" in existing: i += 1 fresh_list = f"l{i}" # set approach i = 1 while f"l{i}" in existing_set: i += 1 fresh_set = f"l{i}" ok = fresh_list == fresh_set record(f"0003-intg: dense collision fresh name agrees list vs set n={n}", ok, f"list={fresh_list} set={fresh_set}") self.assertEqual(fresh_list, fresh_set) def test_no_collision_returns_first(self): """Empty existing: fresh name is the first candidate.""" existing = [] existing_set = set() i = 1 while f"l{i}" in existing_set: i += 1 fresh = f"l{i}" ok = fresh == "l1" record("0003-intg: empty existing returns first candidate l1", ok, f"fresh={fresh}") self.assertEqual(fresh, "l1") class TestLean40003FunctionalComplexityGate(unittest.TestCase): """lean4-0003 Complexity gate: O(N) fixed path well under O(N^2) threshold.""" def _time_defective(self, n, trials=3): times = [] for _ in range(trials): t0 = time.perf_counter() _fresh_name_defective(n) times.append(time.perf_counter() - t0) return min(times) def _time_fixed(self, n, trials=3): times = [] for _ in range(trials): t0 = time.perf_counter() _fresh_name_fixed(n) times.append(time.perf_counter() - t0) return min(times) def test_fixed_n1000_under_5ms(self): """Fixed path N=1000 must complete in <5ms (benchmark: 0.095ms).""" t = self._time_fixed(1000) * 1000 ok = t < 5.0 record(f"0003-gate: fixed N=1000 <5ms", ok, f"{t:.3f}ms") self.assertLess(t, 5.0, f"fixed N=1000 took {t:.3f}ms, expected <5ms") def test_fixed_scaling_linear(self): """Fixed path: time(N=4000) / time(N=1000) must be <10x (O(N) not O(N^2)). Using 4x N step with 10x ratio budget to accommodate Python timer variance at these sub-millisecond timescales. O(N^2) would give ~16x ratio.""" t1 = self._time_fixed(1000) t2 = self._time_fixed(4000) ratio = t2 / t1 if t1 > 0 else 0 ok = ratio < 10.0 record(f"0003-gate: fixed k-scaling <10x for 4x N step (got {ratio:.2f}x)", ok, f"t(1000)={t1*1000:.3f}ms t(4000)={t2*1000:.3f}ms") self.assertLess(ratio, 10.0, f"fixed time ratio {ratio:.2f}x >= 10x -- O(N^2) regression detected") def test_defective_scaling_quadratic(self): """Defective path: time(N=2000) / time(N=1000) must be >3x (confirms O(N^2)).""" t1 = self._time_defective(1000) t2 = self._time_defective(2000) ratio = t2 / t1 if t1 > 0 else 0 ok = ratio > 3.0 record(f"0003-gate: defective scaling >3x confirms O(N^2) (got {ratio:.2f}x)", ok, f"t(1000)={t1*1000:.3f}ms t(2000)={t2*1000:.3f}ms") self.assertGreater(ratio, 3.0, f"defective ratio {ratio:.2f}x not >3x -- benchmark model may be wrong") def test_fixed_faster_than_defective(self): """Fixed path N=1000 must be faster than defective N=1000.""" td = self._time_defective(1000) * 1000 tf = self._time_fixed(1000) * 1000 ok = tf < td record(f"0003-gate: fixed faster than defective N=1000", ok, f"defective={td:.3f}ms fixed={tf:.3f}ms speedup={td/tf:.1f}x") self.assertLess(tf, td) # --------------------------------------------------------------------------- # lean4-0004: double-checked locking race in ir_interpreter lookup_symbol # (CWE-362 -- correctness model, no timing gate applicable) # --------------------------------------------------------------------------- class _SymbolCache: """Python model of m_symbol_cache (instance-local, no global lock needed).""" def __init__(self): self._cache = {} def get(self, key): return self._cache.get(key) def insert(self, key, value): self._cache[key] = value class _NativeSymbolCache: """Python model of g_native_symbol_cache (shared across threads, needs lock).""" def __init__(self, symbols): self._symbols = dict(symbols) self._lock = threading.Lock() def find_under_lock(self, key): """Single exclusive acquisition -- models single unique_lock fix.""" with self._lock: return self._symbols.get(key) def _lookup_symbol_defective(fn, local_cache, global_cache, global_lock): """ Models the double-checked locking defect: 1. Check global without lock (unsafe read). 2. Acquire shared lock. 3. Release shared lock. 4. Acquire unique lock. 5. Check again and insert. Race window between steps 3 and 4 allows another thread to modify cache. Python simulation: skips the unsafe read step but models the upgrade pattern. """ e = local_cache.get(fn) if e is not None: return e # Simulated unsafe pre-check (no lock) -- data race in C++ on weak-order arch. # Then: shared lock, release, unique lock -- upgrade race window. with global_lock: # shared lock (simulated as exclusive in Python) ne = global_cache._symbols.get(fn) if ne is not None: entry = ("decl:" + fn, ne) local_cache.insert(fn, entry) return entry # Re-acquire (models unlock + unique_lock upgrade -- race window here) with global_lock: ne = global_cache._symbols.get(fn) if ne is not None: entry = ("decl:" + fn, ne) local_cache.insert(fn, entry) return entry entry = ("decl:" + fn, None) global_cache._symbols[fn] = None local_cache.insert(fn, entry) return entry def _lookup_symbol_fixed(fn, local_cache, global_cache): """ Models the single unique_lock fix: 1. Check local cache (no lock needed -- instance-local). 2. Acquire one exclusive lock for all global cache access. 3. Release lock. No race window. """ e = local_cache.get(fn) if e is not None: return e # Single exclusive lock acquisition. ne = global_cache.find_under_lock(fn) if ne is not None: entry = ("decl:" + fn, ne) local_cache.insert(fn, entry) return entry entry = ("decl:" + fn, None) local_cache.insert(fn, entry) return entry class TestLean40004Unit(unittest.TestCase): """lean4-0004 Unit: lookup_symbol correctness -- local hit, global hit, miss.""" def _make_caches(self, natives): lock = threading.Lock() local = _SymbolCache() native = _NativeSymbolCache(natives) return local, native, lock def test_local_cache_hit(self): local, native, lock = self._make_caches({"foo": 0xDEAD}) local.insert("foo", ("decl:foo", 0xDEAD)) result = _lookup_symbol_fixed("foo", local, native) ok = result == ("decl:foo", 0xDEAD) record("0004-unit: local cache hit returns cached entry", ok, f"result={result}") self.assertEqual(result, ("decl:foo", 0xDEAD)) def test_global_cache_hit(self): local, native, lock = self._make_caches({"bar": 0xBEEF}) result = _lookup_symbol_fixed("bar", local, native) ok = result == ("decl:bar", 0xBEEF) record("0004-unit: global cache hit returns correct entry", ok, f"result={result}") self.assertEqual(result, ("decl:bar", 0xBEEF)) def test_cache_miss(self): local, native, lock = self._make_caches({}) result = _lookup_symbol_fixed("baz", local, native) ok = result == ("decl:baz", None) record("0004-unit: cache miss returns entry with null native ptr", ok, f"result={result}") self.assertEqual(result, ("decl:baz", None)) def test_local_populated_after_global_hit(self): """After a global hit, subsequent lookup is served from local cache.""" local, native, lock = self._make_caches({"qux": 0x1234}) _lookup_symbol_fixed("qux", local, native) # Remove from native to confirm second call uses local cache native._symbols.clear() result = _lookup_symbol_fixed("qux", local, native) ok = result == ("decl:qux", 0x1234) record("0004-unit: local cache populated after global hit", ok, f"result={result}") self.assertEqual(result, ("decl:qux", 0x1234)) def test_fixed_and_defective_agree_single_thread(self): """Fixed and defective paths return the same result in single-threaded use.""" natives = {"sym_a": 0xAABB, "sym_b": 0xCCDD} lock = threading.Lock() local_d = _SymbolCache() native_d = _NativeSymbolCache(natives) local_f = _SymbolCache() native_f = _NativeSymbolCache(natives) for fn in ["sym_a", "sym_b", "sym_missing"]: rd = _lookup_symbol_defective(fn, local_d, native_d, lock) rf = _lookup_symbol_fixed(fn, local_f, native_f) if rd != rf: record(f"0004-unit: fixed/defective agree single-thread fn={fn}", False, f"defective={rd} fixed={rf}") self.assertEqual(rd, rf) record("0004-unit: fixed and defective agree single-thread (3 symbols)", True) class TestLean40004Integration(unittest.TestCase): """lean4-0004 Integration: concurrent lookup correctness under fixed path.""" def test_concurrent_lookups_all_succeed(self): """N threads each looking up same symbol all get consistent result.""" n_threads = 8 symbols = {f"sym_{i}": i * 0x100 for i in range(20)} native = _NativeSymbolCache(symbols) results = {} errors = [] lock = threading.Lock() def worker(fn): local = _SymbolCache() try: result = _lookup_symbol_fixed(fn, local, native) with lock: results[fn] = result except Exception as e: with lock: errors.append(str(e)) threads = [threading.Thread(target=worker, args=(f"sym_{i}",)) for i in range(n_threads)] for t in threads: t.start() for t in threads: t.join() ok = len(errors) == 0 and len(results) == n_threads record(f"0004-intg: {n_threads} concurrent lookups all succeed", ok, f"results={len(results)} errors={len(errors)}") self.assertEqual(len(errors), 0) self.assertEqual(len(results), n_threads) def test_no_result_corruption_under_contention(self): """Multiple threads looking up the same key all get the same value.""" native = _NativeSymbolCache({"shared_sym": 0xFACE}) results = [] lock_r = threading.Lock() def worker(): local = _SymbolCache() r = _lookup_symbol_fixed("shared_sym", local, native) with lock_r: results.append(r) threads = [threading.Thread(target=worker) for _ in range(16)] for t in threads: t.start() for t in threads: t.join() ok = all(r == ("decl:shared_sym", 0xFACE) for r in results) record(f"0004-intg: 16 threads on shared_sym all get identical result", ok, f"unique_results={len(set(map(str, results)))}") self.assertTrue(ok) def test_miss_consistent_under_contention(self): """Multiple threads looking up a missing key all get the null-native entry.""" native = _NativeSymbolCache({}) results = [] lock_r = threading.Lock() def worker(): local = _SymbolCache() r = _lookup_symbol_fixed("no_such_sym", local, native) with lock_r: results.append(r) threads = [threading.Thread(target=worker) for _ in range(8)] for t in threads: t.start() for t in threads: t.join() ok = all(r == ("decl:no_such_sym", None) for r in results) record(f"0004-intg: 8 threads on missing symbol all get null-native entry", ok) self.assertTrue(ok) class TestLean40004FunctionalComplexityGate(unittest.TestCase): """lean4-0004 Correctness gate: fixed path produces no lost lookups under concurrency.""" def test_high_concurrency_no_lost_results(self): """32 threads, 10 symbols each: all 320 lookups succeed with correct values.""" n_threads = 32 n_symbols = 10 symbols = {f"fn_{i}": i for i in range(n_symbols)} native = _NativeSymbolCache(symbols) total_expected = n_threads * n_symbols results = [] lock_r = threading.Lock() def worker(): local = _SymbolCache() for i in range(n_symbols): r = _lookup_symbol_fixed(f"fn_{i}", local, native) with lock_r: results.append(r) threads = [threading.Thread(target=worker) for _ in range(n_threads)] for t in threads: t.start() for t in threads: t.join() ok = len(results) == total_expected and all(r is not None for r in results) record(f"0004-gate: 32 threads x 10 symbols = {total_expected} lookups, none lost", ok, f"got={len(results)}") self.assertEqual(len(results), total_expected) def test_correctness_values_under_concurrency(self): """All thread-local results match expected native values.""" n_threads = 16 symbols = {f"fn_{i}": i * 7 for i in range(5)} native = _NativeSymbolCache(symbols) mismatches = [] lock_r = threading.Lock() def worker(tid): local = _SymbolCache() for key, expected_native in symbols.items(): r = _lookup_symbol_fixed(key, local, native) if r != ("decl:" + key, expected_native): with lock_r: mismatches.append((tid, key, r)) threads = [threading.Thread(target=worker, args=(i,)) for i in range(n_threads)] for t in threads: t.start() for t in threads: t.join() ok = len(mismatches) == 0 record(f"0004-gate: {n_threads} threads correctness, no value mismatches", ok, f"mismatches={len(mismatches)}") self.assertEqual(len(mismatches), 0) # --------------------------------------------------------------------------- # lean4-0005: Lake job registry IO.Ref race -> Std.Mutex # (CWE-362 -- concurrent push correctness model) # --------------------------------------------------------------------------- class _JobRefDefective: """Models IO.Ref (Array OpaqueJob) -- not atomic: concurrent modify loses entries.""" def __init__(self): self._array = [] # No lock -- models bare IO.Ref def push(self, job): # Simulate non-atomic read-modify-write: read, compute, write current = list(self._array) # Yield opportunity for interleaving (Python GIL limits this but models intent) current.append(job) self._array = current def get(self): return list(self._array) class _JobMutexFixed: """Models Std.Mutex (Array OpaqueJob) -- atomic modify via lock.""" def __init__(self): self._array = [] self._lock = threading.Lock() def push_atomically(self, job): with self._lock: self._array.append(job) def get(self): with self._lock: return list(self._array) class TestLean40005Unit(unittest.TestCase): """lean4-0005 Unit: single-threaded push correctness for both Ref and Mutex models.""" def test_ref_push_single_thread(self): ref = _JobRefDefective() for i in range(10): ref.push(f"job_{i}") ok = len(ref.get()) == 10 record("0005-unit: IO.Ref push 10 jobs single-thread", ok, f"count={len(ref.get())}") self.assertEqual(len(ref.get()), 10) def test_mutex_push_single_thread(self): mx = _JobMutexFixed() for i in range(10): mx.push_atomically(f"job_{i}") ok = len(mx.get()) == 10 record("0005-unit: Std.Mutex push 10 jobs single-thread", ok, f"count={len(mx.get())}") self.assertEqual(len(mx.get()), 10) def test_mutex_ordering_preserved(self): mx = _JobMutexFixed() for i in range(5): mx.push_atomically(f"job_{i}") jobs = mx.get() ok = jobs == [f"job_{i}" for i in range(5)] record("0005-unit: Std.Mutex insertion order preserved (5 jobs)", ok, f"jobs={jobs}") self.assertEqual(jobs, [f"job_{i}" for i in range(5)]) def test_mutex_empty_initial(self): mx = _JobMutexFixed() ok = mx.get() == [] record("0005-unit: Std.Mutex initial state is empty", ok) self.assertEqual(mx.get(), []) def test_mutex_get_returns_copy(self): mx = _JobMutexFixed() mx.push_atomically("job_0") snapshot = mx.get() snapshot.append("external_modification") ok = len(mx.get()) == 1 # internal state unaffected record("0005-unit: Std.Mutex.get returns copy, not reference", ok) self.assertEqual(len(mx.get()), 1) class TestLean40005Integration(unittest.TestCase): """lean4-0005 Integration: mutex correctness under concurrent push.""" def test_mutex_no_lost_registrations(self): """N concurrent threads each push M jobs -- total must be N*M.""" n_threads = 8 n_jobs_each = 25 mx = _JobMutexFixed() def worker(tid): for j in range(n_jobs_each): mx.push_atomically(f"t{tid}_j{j}") threads = [threading.Thread(target=worker, args=(i,)) for i in range(n_threads)] for t in threads: t.start() for t in threads: t.join() total = len(mx.get()) expected = n_threads * n_jobs_each ok = total == expected record(f"0005-intg: {n_threads} threads x {n_jobs_each} jobs = {expected}, got {total}", ok) self.assertEqual(total, expected) def test_mutex_all_job_ids_present(self): """Every pushed job ID appears exactly once in the registry.""" n_threads = 4 n_jobs_each = 50 mx = _JobMutexFixed() expected_ids = set() def worker(tid): for j in range(n_jobs_each): jid = f"t{tid}_j{j}" expected_ids.add(jid) mx.push_atomically(jid) threads = [threading.Thread(target=worker, args=(i,)) for i in range(n_threads)] for t in threads: t.start() for t in threads: t.join() actual_ids = set(mx.get()) ok = actual_ids == expected_ids record(f"0005-intg: all {len(expected_ids)} job IDs present exactly once", ok, f"missing={len(expected_ids - actual_ids)} extra={len(actual_ids - expected_ids)}") self.assertEqual(actual_ids, expected_ids) def test_mutex_modifyget_atomic(self): """Simulates poll() modifyGet: drain returns exact count, none double-counted.""" mx = _JobMutexFixed() for i in range(20): mx.push_atomically(f"job_{i}") # Atomic drain with mx._lock: drained = list(mx._array) mx._array = [] ok = len(drained) == 20 and len(mx.get()) == 0 record(f"0005-intg: atomic drain returns all 20 jobs, leaves queue empty", ok, f"drained={len(drained)} remaining={len(mx.get())}") self.assertEqual(len(drained), 20) self.assertEqual(len(mx.get()), 0) def test_mutex_high_concurrency(self): """32 threads pushing simultaneously, no registrations lost.""" n_threads = 32 mx = _JobMutexFixed() barrier = threading.Barrier(n_threads) def worker(tid): barrier.wait() # maximize contention mx.push_atomically(f"job_{tid}") threads = [threading.Thread(target=worker, args=(i,)) for i in range(n_threads)] for t in threads: t.start() for t in threads: t.join() total = len(mx.get()) ok = total == n_threads record(f"0005-intg: {n_threads}-way concurrent push, all registered", ok, f"expected={n_threads} got={total}") self.assertEqual(total, n_threads) class TestLean40005FunctionalComplexityGate(unittest.TestCase): """lean4-0005 Correctness gate: zero lost registrations across stress run.""" def test_stress_no_lost_jobs(self): """64 threads x 100 jobs = 6400 total, none lost under Std.Mutex model.""" n_threads = 64 n_jobs = 100 mx = _JobMutexFixed() barrier = threading.Barrier(n_threads) def worker(tid): barrier.wait() for j in range(n_jobs): mx.push_atomically((tid, j)) threads = [threading.Thread(target=worker, args=(i,)) for i in range(n_threads)] for t in threads: t.start() for t in threads: t.join() total = len(mx.get()) expected = n_threads * n_jobs ok = total == expected record(f"0005-gate: stress {n_threads}x{n_jobs}={expected} jobs, zero lost", ok, f"got={total}") self.assertEqual(total, expected) def test_no_duplicate_registrations(self): """Each job ID appears exactly once (no double-counting from non-atomic state).""" n_threads = 16 n_jobs = 50 mx = _JobMutexFixed() def worker(tid): for j in range(n_jobs): mx.push_atomically(f"{tid}:{j}") threads = [threading.Thread(target=worker, args=(i,)) for i in range(n_threads)] for t in threads: t.start() for t in threads: t.join() all_jobs = mx.get() ok = len(all_jobs) == len(set(all_jobs)) record(f"0005-gate: no duplicate job IDs in registry", ok, f"total={len(all_jobs)} unique={len(set(all_jobs))}") self.assertEqual(len(all_jobs), len(set(all_jobs))) # --------------------------------------------------------------------------- # lean4-0006: g_opts thread-local not reset at task boundaries # (CWE-668 -- isolation correctness model) # --------------------------------------------------------------------------- class _TraceContextDefective: """ Models LEAN_THREAD_PTR(g_opts) without reset. Thread-local state persists across simulated task boundaries. Python uses threading.local() to model thread-local storage. """ _tls = threading.local() @classmethod def set_opts(cls, opts): cls._tls.g_opts = opts @classmethod def get_opts(cls): return getattr(cls._tls, "g_opts", None) @classmethod def reset_for_task(cls): # Defective: does NOT reset g_opts. pass class _TraceContextFixed: """ Models LEAN_THREAD_PTR(g_opts) WITH register_thread_local_reset_fn. Thread-local state cleared to None at each task boundary. """ _tls = threading.local() @classmethod def set_opts(cls, opts): cls._tls.g_opts = opts @classmethod def get_opts(cls): return getattr(cls._tls, "g_opts", None) @classmethod def reset_for_task(cls): # Fixed: reset_thread_local() calls registered fn: g_opts = nullptr cls._tls.g_opts = None class TestLean40006Unit(unittest.TestCase): """lean4-0006 Unit: thread-local reset behavior.""" def test_defective_leaks_opts(self): """Defective: opts set in task 1 visible in task 2 on same thread.""" _TraceContextDefective.set_opts({"trace": True, "level": 3}) _TraceContextDefective.reset_for_task() # no-op in defective leaked = _TraceContextDefective.get_opts() ok = leaked is not None # leak confirmed record("0006-unit: defective leaks opts across task boundary", ok, f"leaked={leaked}") self.assertIsNotNone(leaked) def test_fixed_clears_opts(self): """Fixed: opts set in task 1 cleared after reset_for_task.""" _TraceContextFixed.set_opts({"trace": True, "level": 3}) _TraceContextFixed.reset_for_task() after = _TraceContextFixed.get_opts() ok = after is None record("0006-unit: fixed clears opts at task boundary", ok, f"after_reset={after}") self.assertIsNone(after) def test_fixed_initial_state_is_none(self): results = [] def check(): results.append(_TraceContextFixed.get_opts()) t = threading.Thread(target=check) t.start() t.join() ok = results[0] is None record("0006-unit: fixed initial g_opts is None on fresh thread", ok) self.assertIsNone(results[0]) def test_fixed_new_task_starts_clean(self): """Simulate task 1 sets opts, reset, task 2 reads None.""" _TraceContextFixed.set_opts({"important": "secret_trace"}) _TraceContextFixed.reset_for_task() # Simulate task 2 on same thread opts_task2 = _TraceContextFixed.get_opts() ok = opts_task2 is None record("0006-unit: task 2 starts with None after reset", ok, f"opts_task2={opts_task2}") self.assertIsNone(opts_task2) def test_set_after_reset_works(self): """After reset, new task can set its own opts without interference.""" _TraceContextFixed.set_opts({"from_task_1": True}) _TraceContextFixed.reset_for_task() _TraceContextFixed.set_opts({"from_task_2": True}) ok = _TraceContextFixed.get_opts() == {"from_task_2": True} record("0006-unit: task 2 can set its own opts after reset", ok) self.assertEqual(_TraceContextFixed.get_opts(), {"from_task_2": True}) class TestLean40006Integration(unittest.TestCase): """lean4-0006 Integration: no cross-task trace leakage under thread-pool reuse.""" def test_thread_reuse_isolation_fixed(self): """ Simulate thread pool: same thread runs task 1 (sets opts), then task 2. Fixed: task 2 sees None. Defective: task 2 sees task 1's opts. """ task2_opts = [] def run_tasks(): # Task 1 _TraceContextFixed.set_opts({"task": 1, "verbose": True}) # Simulate task boundary -- reset_thread_local() fires _TraceContextFixed.reset_for_task() # Task 2 task2_opts.append(_TraceContextFixed.get_opts()) t = threading.Thread(target=run_tasks) t.start() t.join() ok = task2_opts[0] is None record("0006-intg: thread reuse -- task 2 opts isolated (fixed)", ok, f"task2_opts={task2_opts[0]}") self.assertIsNone(task2_opts[0]) def test_multiple_tasks_on_same_thread_all_isolated(self): """Each of 5 sequential tasks on same thread sees None at start.""" seen_opts = [] def run_five_tasks(): for i in range(5): _TraceContextFixed.reset_for_task() # boundary before each task seen_opts.append(_TraceContextFixed.get_opts()) _TraceContextFixed.set_opts({"task": i}) t = threading.Thread(target=run_five_tasks) t.start() t.join() ok = all(o is None for o in seen_opts) record(f"0006-intg: 5 sequential tasks all start with None", ok, f"seen={seen_opts}") self.assertTrue(ok) def test_parallel_threads_no_cross_contamination(self): """ N threads run concurrently, each sets its own opts. Each thread confirms it reads back its own opts, not another thread's. """ n_threads = 8 mismatches = [] lock_r = threading.Lock() def worker(tid): _TraceContextFixed.reset_for_task() expected = {"thread": tid} _TraceContextFixed.set_opts(expected) actual = _TraceContextFixed.get_opts() if actual != expected: with lock_r: mismatches.append((tid, actual)) threads = [threading.Thread(target=worker, args=(i,)) for i in range(n_threads)] for t in threads: t.start() for t in threads: t.join() ok = len(mismatches) == 0 record(f"0006-intg: {n_threads} parallel threads, no cross-contamination", ok, f"mismatches={len(mismatches)}") self.assertEqual(len(mismatches), 0) def test_defective_shows_leak_on_reuse(self): """Confirm the defective model DOES leak (validates our test is meaningful).""" leaked_opts = [] def run_tasks(): _TraceContextDefective.set_opts({"secret": "from_task_1"}) _TraceContextDefective.reset_for_task() # no-op leaked_opts.append(_TraceContextDefective.get_opts()) t = threading.Thread(target=run_tasks) t.start() t.join() ok = leaked_opts[0] is not None # defective SHOULD leak record("0006-intg: defective model confirms leak (validates test sensitivity)", ok, f"leaked={leaked_opts[0]}") self.assertIsNotNone(leaked_opts[0]) class TestLean40006FunctionalComplexityGate(unittest.TestCase): """lean4-0006 Correctness gate: zero leakage events in stress run.""" def test_no_leakage_stress(self): """64 threads, each running 10 simulated tasks: zero opts leaked.""" n_threads = 64 n_tasks = 10 leaks = [] lock_r = threading.Lock() def worker(tid): for task in range(n_tasks): _TraceContextFixed.reset_for_task() start_opts = _TraceContextFixed.get_opts() if start_opts is not None: with lock_r: leaks.append((tid, task, start_opts)) _TraceContextFixed.set_opts({"tid": tid, "task": task}) threads = [threading.Thread(target=worker, args=(i,)) for i in range(n_threads)] for t in threads: t.start() for t in threads: t.join() ok = len(leaks) == 0 record(f"0006-gate: {n_threads} threads x {n_tasks} tasks, zero leakage events", ok, f"leaks={len(leaks)}") self.assertEqual(len(leaks), 0) # --------------------------------------------------------------------------- # lean4-0007: runtime/process.cpp env var inheritance O(N^2) -> O(N) # --------------------------------------------------------------------------- def _envvar_inherit_defective(inherited_env, override_keys_list): """ O(N*M): for each of N inherited env vars, construct a string from char* range and call new_env_vars.count() -- modeled as list membership O(M). N = len(inherited_env), M = len(override_keys_list). """ result = [] for key in inherited_env: # Models: new_env_vars.count({key_begin, key_end}) -- O(M) list scan if key not in override_keys_list: # O(M) result.append(key) return result def _envvar_inherit_fixed(inherited_env, override_keys_list): """ O(N+M): pre-build unordered_set of override keys, then O(1) lookup per inherited var. """ override_set = set(override_keys_list) # O(M) build once result = [] for key in inherited_env: if key not in override_set: # O(1) result.append(key) return result class TestLean40007Unit(unittest.TestCase): """lean4-0007 Unit: env var inheritance correctness, list vs set lookup.""" def test_no_override_all_inherited(self): env = ["PATH", "HOME", "USER"] result = _envvar_inherit_fixed(env, []) ok = result == env record("0007-unit: no overrides -- all env vars inherited", ok, f"result={result}") self.assertEqual(result, env) def test_all_overridden_none_inherited(self): env = ["PATH", "HOME", "USER"] result = _envvar_inherit_fixed(env, ["PATH", "HOME", "USER"]) ok = result == [] record("0007-unit: all overridden -- no env vars inherited", ok, f"result={result}") self.assertEqual(result, []) def test_partial_override(self): env = ["PATH", "HOME", "USER", "TERM"] overrides = ["HOME", "TERM"] result = _envvar_inherit_fixed(env, overrides) ok = set(result) == {"PATH", "USER"} record("0007-unit: partial override -- correct subset inherited", ok, f"result={result}") self.assertEqual(set(result), {"PATH", "USER"}) def test_list_and_set_agree_small(self): env = [f"VAR_{i}" for i in range(20)] overrides = [f"VAR_{i}" for i in range(0, 20, 2)] # even-numbered overridden list_result = _envvar_inherit_defective(env, overrides) set_result = _envvar_inherit_fixed(env, overrides) ok = list_result == set_result record("0007-unit: list and set approaches return same result (20 vars)", ok, f"list={len(list_result)} set={len(set_result)}") self.assertEqual(list_result, set_result) def test_empty_env_empty_result(self): result = _envvar_inherit_fixed([], ["OVERRIDE"]) ok = result == [] record("0007-unit: empty inherited env returns empty", ok) self.assertEqual(result, []) class TestLean40007Integration(unittest.TestCase): """lean4-0007 Integration: correctness at medium scale env var sets.""" def test_medium_env_list_vs_set_agree(self): n = 500 env = [f"VAR_{i}" for i in range(n)] overrides = [f"VAR_{i}" for i in range(0, n, 3)] # every 3rd overridden list_result = _envvar_inherit_defective(env, overrides) set_result = _envvar_inherit_fixed(env, overrides) ok = list_result == set_result record(f"0007-intg: N={n} env vars, list/set results agree", ok, f"inherited={len(set_result)} overridden={len(overrides)}") self.assertEqual(list_result, set_result) def test_no_duplicates_in_result(self): n = 200 env = [f"VAR_{i}" for i in range(n)] overrides = [f"VAR_{i}" for i in range(50)] result = _envvar_inherit_fixed(env, overrides) ok = len(result) == len(set(result)) record(f"0007-intg: no duplicate env vars in inherited result N={n}", ok, f"total={len(result)} unique={len(set(result))}") self.assertEqual(len(result), len(set(result))) def test_override_set_not_in_result(self): n = 300 env = [f"VAR_{i}" for i in range(n)] overrides_set = {f"VAR_{i}" for i in range(100, 200)} overrides_list = list(overrides_set) result = _envvar_inherit_fixed(env, overrides_list) leaked = [k for k in result if k in overrides_set] ok = len(leaked) == 0 record(f"0007-intg: no overridden keys appear in inherited result N={n}", ok, f"leaked={len(leaked)}") self.assertEqual(len(leaked), 0) def test_non_overridden_all_present(self): n = 300 env = [f"VAR_{i}" for i in range(n)] overrides = [f"VAR_{i}" for i in range(100)] result_set = set(_envvar_inherit_fixed(env, overrides)) expected = {f"VAR_{i}" for i in range(100, n)} ok = result_set == expected record(f"0007-intg: all non-overridden vars present in result N={n}", ok, f"expected={len(expected)} got={len(result_set)}") self.assertEqual(result_set, expected) class TestLean40007FunctionalComplexityGate(unittest.TestCase): """lean4-0007 Complexity gate: O(N+M) fixed path well under O(N*M) threshold.""" def _time_defective(self, n, m, trials=3): env = [f"VAR_{i}" for i in range(n)] overrides = [f"OVER_{i}" for i in range(m)] # disjoint: scan goes full length times = [] for _ in range(trials): t0 = time.perf_counter() _envvar_inherit_defective(env, overrides) times.append(time.perf_counter() - t0) return min(times) def _time_fixed(self, n, m, trials=3): env = [f"VAR_{i}" for i in range(n)] overrides = [f"OVER_{i}" for i in range(m)] times = [] for _ in range(trials): t0 = time.perf_counter() _envvar_inherit_fixed(env, overrides) times.append(time.perf_counter() - t0) return min(times) def test_fixed_n500_under_1ms(self): """Fixed path N=500, M=500 must complete in <1ms (patch gate).""" t = self._time_fixed(500, 500) * 1000 ok = t < 1.0 record(f"0007-gate: fixed N=500 M=500 <1ms", ok, f"{t:.3f}ms") self.assertLess(t, 1.0, f"fixed N=500 M=500 took {t:.3f}ms, expected <1ms") def test_fixed_scaling_linear(self): """Fixed path: time(N=1000) / time(N=500) must be <3x (O(N+M) not O(N*M)).""" t1 = self._time_fixed(500, 500) t2 = self._time_fixed(1000, 1000) ratio = t2 / t1 if t1 > 0 else 0 ok = ratio < 3.0 record(f"0007-gate: fixed k-scaling <3x (got {ratio:.2f}x)", ok, f"t(500)={t1*1000:.3f}ms t(1000)={t2*1000:.3f}ms") self.assertLess(ratio, 3.0, f"fixed time ratio {ratio:.2f}x >= 3x -- O(N*M) regression detected") def test_defective_scaling_quadratic(self): """Defective path: time(N=M=2000) / time(N=M=500) must be >5x (confirms O(N*M)). 4x N and M step: O(N*M) predicts 16x ratio; O(N+M) predicts 4x ratio. Requiring >5x confirms super-linear growth well above the O(N+M) baseline.""" t1 = self._time_defective(500, 500) t2 = self._time_defective(2000, 2000) ratio = t2 / t1 if t1 > 0 else 0 ok = ratio > 5.0 record(f"0007-gate: defective scaling >5x for 4x N step (got {ratio:.2f}x)", ok, f"t(500)={t1*1000:.3f}ms t(2000)={t2*1000:.3f}ms") self.assertGreater(ratio, 5.0, f"defective ratio {ratio:.2f}x not >5x -- benchmark model may be wrong") def test_fixed_faster_than_defective(self): """Fixed path N=M=1000 must be faster than defective N=M=1000.""" td = self._time_defective(1000, 1000) * 1000 tf = self._time_fixed(1000, 1000) * 1000 ok = tf < td record(f"0007-gate: fixed faster than defective N=M=1000", ok, f"defective={td:.3f}ms fixed={tf:.3f}ms speedup={td/tf:.1f}x") self.assertLess(tf, td) # --------------------------------------------------------------------------- # Runner # --------------------------------------------------------------------------- def main(): print("=" * 70) print("lean4 CWE-407/362/668 patch test suite") print("UNDF-2026-000001259 through UNDF-2026-000001265") print("=" * 70) loader = unittest.TestLoader() suite = unittest.TestSuite() test_classes = [ TestLean40001Unit, TestLean40001Integration, TestLean40001FunctionalComplexityGate, TestLean40002Unit, TestLean40002Integration, TestLean40002FunctionalComplexityGate, TestLean40003Unit, TestLean40003Integration, TestLean40003FunctionalComplexityGate, TestLean40004Unit, TestLean40004Integration, TestLean40004FunctionalComplexityGate, TestLean40005Unit, TestLean40005Integration, TestLean40005FunctionalComplexityGate, TestLean40006Unit, TestLean40006Integration, TestLean40006FunctionalComplexityGate, TestLean40007Unit, TestLean40007Integration, TestLean40007FunctionalComplexityGate, ] for cls in test_classes: print(f"\n--- {cls.__name__} ---") tests = loader.loadTestsFromTestCase(cls) suite.addTests(tests) runner = unittest.TextTestRunner( stream=open("/dev/null", "w"), verbosity=0 ) result = runner.run(tests) print("\n" + "=" * 70) total = len(_results) passed = sum(_results) failed = total - passed print(f"Results: {passed}/{total} passed, {failed} failed") print("=" * 70) if failed > 0: print(f"\nFAIL -- {failed} test(s) failed") sys.exit(1) else: print("\nPASS -- all tests passed") sys.exit(0) if __name__ == "__main__": main()