From c9314c264b4b2dcfd3ae99af99955518f5d89e84 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Mon, 13 Apr 2026 11:03:16 -0400 Subject: [PATCH] =?UTF-8?q?test:=20add=20lean4=20test=20suite=20=E2=80=94?= =?UTF-8?q?=2088=20tests=20across=207=20patches=20(CWE-407=20+=20CWE-362)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Unit, integration, and functional complexity gates for lean4-0001..0007. All 88 tests pass. Complexity gates enforce measured speedups: lean4-0001: N=2000 fixed <20ms (vs 38ms defective, 34x) lean4-0002: K=N=1000 fixed <5ms (vs 14ms defective, 678x) lean4-0003: N=1000 fixed <5ms (vs 20ms defective, 210x) lean4-0005/0006: concurrent correctness (zero lost jobs/leaked context) lean4-0007: N=M=500 fixed <1ms (vs O(N*M) defective) --- defects/lean4/Makefile | 19 + defects/lean4/tests/test-lean4-cwe407.py | 1695 ++++++++++++++++++++++ 2 files changed, 1714 insertions(+) create mode 100644 defects/lean4/Makefile create mode 100644 defects/lean4/tests/test-lean4-cwe407.py diff --git a/defects/lean4/Makefile b/defects/lean4/Makefile new file mode 100644 index 000000000..2b3cee821 --- /dev/null +++ b/defects/lean4/Makefile @@ -0,0 +1,19 @@ +# lean4 patch test + bench runner +# Targets: all test bench clean + +PYTHON := python3 +TEST_FILE := tests/test-lean4-cwe407.py +BENCH_DIR := bench + +.PHONY: all test bench clean + +all: test bench + +test: + $(PYTHON) $(TEST_FILE) + +bench: + $(PYTHON) $(BENCH_DIR)/run_all.py + +clean: + rm -rf tests/__pycache__ bench/__pycache__ __pycache__ diff --git a/defects/lean4/tests/test-lean4-cwe407.py b/defects/lean4/tests/test-lean4-cwe407.py new file mode 100644 index 000000000..fd68e38a7 --- /dev/null +++ b/defects/lean4/tests/test-lean4-cwe407.py @@ -0,0 +1,1695 @@ +#!/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()