Seed all random lineage generation in tests to fix flaky CI failures

Several tests defined a seeded rng but never passed it into
LineageTree.rand_init, Analyze_list, or hide_observation, so those
calls fell back to global unseeded randomness. Occasionally this
produced degenerate data (e.g. all cells eliminated from an estimator,
or optimizer non-convergence on badly conditioned draws) causing
sporadic CI failures. Thread the seeded rng through consistently, and
add an rng parameter to crossval/hide_observation so tests can do the
same.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PVLX72yJpEABz61TPrJ9Ea

Author: Aaron Meyer

lineage/crossval.py

@@ -12,33 +12,35 @@
 exe = ProcessPoolExecutor()
 
 
-def hide_observation(lineages: list, percentage: float) -> list:
+def hide_observation(lineages: list, percentage: float, rng=None) -> list:
     """Taking a list of lineages and the percentage of cells want to be masked, it marks those x% negative."""
+    rng = np.random.default_rng(rng)
     new_lineages = deepcopy(lineages)
     for new_lineage in new_lineages:
         for cell in new_lineage.output_lineage:
-            if bernoulli.rvs(p=percentage, size=1):
+            if bernoulli.rvs(p=percentage, size=1, random_state=rng):
                 # negate the cell observations to mask them
                 cell.obs = [-1 * o for o in cell.obs]
         new_lineage.obs = np.array([cell.obs for cell in new_lineage.output_lineage])
 
     return new_lineages
 
 
-def crossval(train_populations: list, num_states: np.ndarray):
+def crossval(train_populations: list, num_states: np.ndarray, rng=None):
     """Perform cross validation for the experimental data which runs in parallel for all states.
     :param train_populations: the populations after applying hide_observation. This includes the list of list of lineages.
     :param hidden_indexes: is a list of list of np.arrays for each lineage,
     filled with zeros and ones. ones refer to the index of those cells that have been hidden.
     :param hidden_obs: list of list of tuples of observations that have been masked in the train_lineage.
     :param num_states: is a range of states we want to run the cross validation for.
     """
+    rng = np.random.default_rng(rng)
     # fit training data by parallel.
 
     # save the tHMMobj for each number of states that is being run
     LLs = []
     for k in num_states:
-        tHMMobj_list, _, gamma_list = Analyze_list(train_populations, k)
+        tHMMobj_list, _, gamma_list = Analyze_list(train_populations, k, rng=rng)
 
         Logls = 0
         # calculate the log likelihood of hidden observations

lineage/tests/test_BaumWelch.py

@@ -55,6 +55,7 @@ def test_BW(cens, nStates):
         desired_num_cells=num_cells,
         desired_experiment_time=expt_time,
         censor_condition=cens,
+        rng=rng,
     )
     tHMMobj = tHMM([X], num_states=nStates)  # build the tHMM class with X
 

lineage/tests/test_CrossVal.py

@@ -38,7 +38,7 @@ def test_cv(cen):
         for _ in range(4)
     ]
 
-    train_lineages = [hide_observation(complete_lin, 0.25) for complete_lin in complete_lineages]
+    train_lineages = [hide_observation(complete_lin, 0.25, rng=local_rng) for complete_lin in complete_lineages]
 
-    ll = crossval(train_lineages, np.arange(1, 3))
+    ll = crossval(train_lineages, np.arange(1, 3), rng=local_rng)
     assert ll[0] < ll[1]

lineage/tests/test_StateEstimation.py

@@ -35,11 +35,12 @@ def gen():
             2**8,
             censor_condition=censored,
             desired_experiment_time=100,
+            rng=rng,
         )
 
     lineage_gamma = [gen() for _ in range(50)]
     solver_gamma = tHMM(lineage_gamma, 1)  # evaluating for one state
-    fit_list([solver_gamma])
+    fit_list([solver_gamma], rng=rng)
 
     assert solver_gamma.estimate.E[0].dist(E_gamma[0]) < 4.0
 

lineage/tests/test_tHMM.py

@@ -88,7 +88,7 @@ def test_fit_performance():
 def test_small_lineages(sizze, stateNum):
     """To test lineages with 3 cells in them for simple gamma."""
     # test with 2 state model
-    lin = [LineageTree.rand_init(pi, T, E, sizze) for _ in range(2)]
+    lin = [LineageTree.rand_init(pi, T, E, sizze, rng=rng) for _ in range(2)]
 
     _, LL1, _ = Analyze_list([lin], stateNum, rng=rng)
     assert np.all(np.isfinite(LL1))