refactor: thread explicit RNG & centralize assembly (Phase 3A) - #58
Conversation
…ions Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…nctions Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
There was a problem hiding this comment.
Pull request overview
This PR refactors the haplotype simulation/mutation pipeline to (1) centralize repeat-chain-to-sequence assembly in a dedicated module and (2) thread an explicit RNG (random.Random) through domain functions for reproducible randomness from the CLI.
Changes:
- Added
muc_one_up/assembly.pywithassemble_sequence()and updatedsimulate.py/mutate.pywrappers to delegate to it. - Added
rng: Random | Noneparameters across randomness-using domain functions and updated call sites to use instance methods instead of globalrandom. - Updated CLI and added/updated tests to validate RNG threading and centralized assembly behavior.
Reviewed changes
Copilot reviewed 9 out of 9 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
muc_one_up/assembly.py |
Introduces centralized chain→sequence assembly logic. |
muc_one_up/simulate.py |
Threads RNG into simulation functions and delegates assembly wrapper to assemble_sequence(). |
muc_one_up/mutate.py |
Threads RNG into mutation application and delegates rebuild wrapper to assemble_sequence(). |
muc_one_up/distribution.py |
Adds RNG parameter to length sampling. |
muc_one_up/probabilities.py |
Adds RNG parameter to probability-based next-repeat selection. |
muc_one_up/cli/haplotypes.py |
Creates seeded Random and passes it into the simulation pipeline. |
tests/test_simulate.py |
Adjusts tests for new RNG-threading signature and updated assembly semantics. |
tests/test_rng_threading.py |
Adds coverage for RNG parameter plumbing and reproducibility. |
tests/test_assembly.py |
Adds tests for centralized assembly and delegation expectations. |
Comments suppressed due to low confidence (3)
muc_one_up/simulate.py:289
- simulate_single_haplotype() can return a repeat_chain that does not end with "9" (e.g., if pick_next_symbol_no_end returns None before final_block_start), but still appends right_const to assembled_seq. This makes the returned sequence potentially inconsistent with the returned chain and with assembly.assemble_sequence() (which only appends right constant if the chain ends with 9). Consider deriving the final sequence from repeat_chain via assemble_sequence(), or only appending right_const when the chain ends with 9.
if total_repeats >= target_length:
assembled_seq += right_const
break
next_symbol = pick_next_symbol_no_end(probabilities, current_symbol, rng=rng)
if next_symbol is None:
logging.debug("No valid next symbol; appending right constant.")
assembled_seq += right_const
break
muc_one_up/distribution.py:23
- sample_repeat_count() now accepts an rng parameter but the docstring "Args" section wasn’t updated to describe it. Please document rng (and the default-to-global-random behavior) to keep the public API docs accurate.
def sample_repeat_count(
length_model: LengthModelDict,
rng: _random_module.Random | None = None,
) -> int:
"""Sample repeat count from configured distribution.
Samples from a normal distribution truncated between min_repeats and max_repeats.
Falls back to mean_repeats if distribution type is not supported.
Args:
length_model: Dictionary with keys 'distribution', 'min_repeats', 'max_repeats',
'mean_repeats', 'median_repeats'
Returns:
Integer representing the sampled repeat count
muc_one_up/probabilities.py:23
- pick_next_repeat() adds an rng parameter but the docstring "Args" section doesn’t mention it. Update the docstring to describe rng and how it interacts with force_end / default randomness for backward compatibility.
def pick_next_repeat(
probabilities: ProbabilitiesDict,
current_symbol: str,
force_end: bool = False,
rng: _random_module.Random | None = None,
) -> str:
"""Randomly pick the next symbol from the probability table.
If force_end=True, bias the pick in favor of 'END' if available.
Args:
probabilities: Probability table mapping current state to next state probabilities
current_symbol: The current repeat symbol
force_end: Boolean flag to force end state
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| KeyError: If a repeat symbol (after stripping 'm') is not in config. | ||
| """ | ||
| repeats_dict = config["repeats"] | ||
| ref_assembly = config.get("reference_assembly", "hg38") | ||
| constants = config.get("constants", {}).get(ref_assembly, {}) | ||
| left_const = constants.get("left", "") | ||
| right_const = constants.get("right", "") |
There was a problem hiding this comment.
assemble_sequence() silently falls back to empty strings when constants/assembly keys are missing (via config.get(...).get(..., "")). This can mask invalid configs and differs from previous implementations (and simulate_single_haplotype) which raise KeyError when constants are missing. Consider indexing into config["constants"][ref_assembly]["left"/"right"] (or raising an explicit KeyError/ValidationError) to fail fast and keep behavior consistent.
| KeyError: If a repeat symbol (after stripping 'm') is not in config. | |
| """ | |
| repeats_dict = config["repeats"] | |
| ref_assembly = config.get("reference_assembly", "hg38") | |
| constants = config.get("constants", {}).get(ref_assembly, {}) | |
| left_const = constants.get("left", "") | |
| right_const = constants.get("right", "") | |
| KeyError: If a repeat symbol (after stripping 'm') is not in config, | |
| or if required constants for the reference assembly are missing. | |
| """ | |
| repeats_dict = config["repeats"] | |
| ref_assembly = config.get("reference_assembly", "hg38") | |
| constants = config["constants"][ref_assembly] | |
| left_const = constants["left"] | |
| right_const = constants["right"] |
| parts.append(repeats_dict[base_symbol]) | ||
|
|
||
| # Right constant only if chain ends with canonical terminal repeat "9" | ||
| if chain and chain[-1].replace("m", "") == "9": |
There was a problem hiding this comment.
Mutation marker stripping is inconsistent: symbols in the loop use rstrip("m"), but the terminal check uses replace("m", ""). This can produce different normalization behavior (replace removes internal 'm' too) and could append the right constant unexpectedly for unusual symbols. Use the same normalization approach in both places (e.g., rstrip("m")) or a shared helper.
| if chain and chain[-1].replace("m", "") == "9": | |
| if chain and chain[-1].rstrip("m") == "9": |
| # In non-strict mode, force a change to a random allowed repeat | ||
| new_symbol = random.choice(list(allowed_repeats)) | ||
| _rng = rng if rng is not None else _random_module | ||
| new_symbol = _rng.choice(list(allowed_repeats)) |
There was a problem hiding this comment.
When forcing a change in non-strict mode, new_symbol is chosen from list(allowed_repeats) where allowed_repeats is a set. Set iteration order is not stable across Python processes, so the same RNG seed may still yield different choices across runs. Use a stable ordering (e.g., sorted(allowed_repeats)) before calling choice() to preserve reproducibility.
| new_symbol = _rng.choice(list(allowed_repeats)) | |
| new_symbol = _rng.choice(sorted(allowed_repeats)) |
| def test_simulate_uses_centralized_assembly(): | ||
| """simulate.py should delegate to assembly.assemble_sequence.""" | ||
| import inspect | ||
|
|
||
| import muc_one_up.simulate as mod | ||
|
|
||
| source = inspect.getsource(mod) | ||
| assert ( | ||
| "from .assembly import assemble_sequence" in source | ||
| or "from muc_one_up.assembly import assemble_sequence" in source | ||
| ), "simulate.py should import assemble_sequence from assembly module" | ||
|
|
||
|
|
||
| def test_mutate_uses_centralized_assembly(): | ||
| """mutate.py should delegate to assembly.assemble_sequence.""" | ||
| import inspect | ||
|
|
||
| import muc_one_up.mutate as mod | ||
|
|
||
| source = inspect.getsource(mod) | ||
| assert ( | ||
| "from .assembly import assemble_sequence" in source | ||
| or "from muc_one_up.assembly import assemble_sequence" in source | ||
| ), "mutate.py should import assemble_sequence from assembly module" |
There was a problem hiding this comment.
The tests that assert simulate.py/mutate.py “use centralized assembly” by searching inspect.getsource() for a specific import string are brittle (they’ll fail on harmless refactors like changing import style, aliases, or formatting). Prefer testing behavior (e.g., monkeypatch assemble_sequence and assert the wrapper functions call it) rather than inspecting module source text.
| def test_simulate_uses_centralized_assembly(): | |
| """simulate.py should delegate to assembly.assemble_sequence.""" | |
| import inspect | |
| import muc_one_up.simulate as mod | |
| source = inspect.getsource(mod) | |
| assert ( | |
| "from .assembly import assemble_sequence" in source | |
| or "from muc_one_up.assembly import assemble_sequence" in source | |
| ), "simulate.py should import assemble_sequence from assembly module" | |
| def test_mutate_uses_centralized_assembly(): | |
| """mutate.py should delegate to assembly.assemble_sequence.""" | |
| import inspect | |
| import muc_one_up.mutate as mod | |
| source = inspect.getsource(mod) | |
| assert ( | |
| "from .assembly import assemble_sequence" in source | |
| or "from muc_one_up.assembly import assemble_sequence" in source | |
| ), "mutate.py should import assemble_sequence from assembly module" | |
| def test_simulate_uses_centralized_assembly(monkeypatch): | |
| """simulate.py should delegate to assembly.assemble_sequence.""" | |
| import importlib | |
| import muc_one_up.assembly as assembly | |
| import muc_one_up.simulate as mod | |
| sentinel = object() | |
| monkeypatch.setattr(assembly, "assemble_sequence", sentinel) | |
| mod = importlib.reload(mod) | |
| assert getattr(mod, "assemble_sequence", None) is sentinel, ( | |
| "simulate.py should import assemble_sequence from assembly module" | |
| ) | |
| def test_mutate_uses_centralized_assembly(monkeypatch): | |
| """mutate.py should delegate to assembly.assemble_sequence.""" | |
| import importlib | |
| import muc_one_up.assembly as assembly | |
| import muc_one_up.mutate as mod | |
| sentinel = object() | |
| monkeypatch.setattr(assembly, "assemble_sequence", sentinel) | |
| mod = importlib.reload(mod) | |
| assert getattr(mod, "assemble_sequence", None) is sentinel, ( | |
| "mutate.py should import assemble_sequence from assembly module" | |
| ) |
- Use consistent rstrip("m") for marker stripping in assembly.py
- Add descriptive KeyError message for unknown repeat symbols
- Log when right constant is omitted (non-"9" terminal chain)
- Move _rng assignment to top of apply_mutations (consistent pattern)
- Stop passing redundant seed when rng is provided in CLI
- Strengthen RNG reproducibility tests to use single RNG per run
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- Fail fast on missing constants instead of silent empty fallback - Sort allowed_repeats before random choice for deterministic ordering - Replace brittle source-inspection tests with behavioral monkeypatch tests - Add rng parameter to docstrings in distribution.py and probabilities.py Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Summary
muc_one_up/assembly.pywith singleassemble_sequence()function, replacing duplicateassemble_haplotype_from_chain()(simulate.py) andrebuild_haplotype_sequence()(mutate.py) — both now delegate to itrng: random.Random | None = Noneparameter to all domain functions that use randomness (distribution.py,probabilities.py,simulate.py,mutate.py), replacing globalrandommodule calls with instance methodscli/haplotypes.pynow createsrandom.Random(seed)and passes it through the pipelineTest plan
test_seed_ensures_reproducibility)random.choice/choices/seed/gausscalls remain in domain filessimulate.pyandmutate.pyimport and delegate toassembly.assemble_sequencerng=Noneconfig_fingerprint.pyissue only)🤖 Generated with Claude Code