| """Focused static contract for the official captured continuation runner.""" |
|
|
| from pathlib import Path |
|
|
| import pytest |
| import torch |
|
|
| from music3lab.editing.music3_continuation import ( |
| BASE_ID, |
| AppendedTrajectory, |
| CapturedFlowRenderer, |
| RebuiltCache, |
| generate_fresh_split, |
| load_continuation_config, |
| ) |
|
|
|
|
| ROOT = Path(__file__).parents[1] |
|
|
|
|
| def test_frozen_continuation_config_is_capture_only(): |
| loaded = load_continuation_config(ROOT / "configs/continuation-tier-a.yaml") |
| assert loaded.config.expected_base_id == BASE_ID |
| assert loaded.config.append_frames == 25 |
| assert loaded.config.equal_power_overlap_samples == 11008 |
| assert sum(case.continue_from_cache for case in loaded.config.cases) == 1 |
| assert {case.source_run_id for case in loaded.config.cases} == { |
| "b0ee35156dad797b7d120419782d5c1d8847f24ae8c5c3aa3d0bd2e1dd7baa6d", |
| "2224b993a75741fb7391b545cb8687c14a131aec4568925bec3dd565ca84a704", |
| } |
|
|
|
|
| def test_official_chunk_layout_static_contract(): |
| assert CapturedFlowRenderer.chunk_starts(25) == [0] |
| assert CapturedFlowRenderer.chunk_starts(200) == [0] |
| assert CapturedFlowRenderer.chunk_starts(300) == [0, 100] |
| assert CapturedFlowRenderer.chunk_starts(325) == [0, 100, 200] |
|
|
|
|
| def test_runner_matches_authenticated_phase0_execution_policy(): |
| source = (ROOT / "src/music3lab/editing/music3_continuation.py").read_text() |
| runner = source.split("def run_captured_continuation(", 1)[1].split( |
| "\ndef ", 1 |
| )[0] |
| assert "torch.use_deterministic_algorithms(False)" in runner |
| assert "torch.use_deterministic_algorithms(True)" not in runner |
|
|
|
|
| def test_fresh_split_reuses_one_generator_and_preserves_rng_order(): |
| class FakeForcer: |
| def __init__(self): |
| self.generator_ids = [] |
|
|
| @staticmethod |
| def _trajectory(values, *, sequence_length, pending, resume=None): |
| tokens = values.reshape(25, 8) |
| return AppendedTrajectory( |
| token_rows=tokens, |
| fused_hidden=tokens[:, :2].T.unsqueeze(0).float(), |
| cache=RebuiltCache( |
| past_key_values=None, |
| last_hidden=torch.empty(0), |
| sequence_length=sequence_length, |
| pending_feedback_row=tokens[-1] if pending else None, |
| ), |
| termination_reason="max_frames", |
| resume_from_sequence_length=(None if resume is None else resume[0]), |
| resume_entry_sequence_length=(None if resume is None else resume[1]), |
| ) |
|
|
| def generate_fresh_prefix(self, *, requested_frames, generator): |
| assert requested_frames == 25 |
| self.generator_ids.append(id(generator)) |
| values = torch.randint(0, 1000, (200,), generator=generator) |
| return ( |
| torch.arange(8), |
| self._trajectory(values, sequence_length=129, pending=True), |
| 104, |
| "0" * 64, |
| ) |
|
|
| def resume(self, cache, *, requested_frames, generator): |
| assert cache.sequence_length == 129 |
| assert requested_frames == 25 |
| self.generator_ids.append(id(generator)) |
| values = torch.randint(0, 1000, (200,), generator=generator) |
| return self._trajectory( |
| values, |
| sequence_length=154, |
| pending=True, |
| resume=(129, 130), |
| ) |
|
|
| forcer = FakeForcer() |
| generator = torch.Generator().manual_seed(7007) |
| result = generate_fresh_split( |
| forcer, |
| requested_frames=25, |
| generator=generator, |
| ) |
| reference = torch.randint( |
| 0, |
| 1000, |
| (400,), |
| generator=torch.Generator().manual_seed(7007), |
| ).reshape(50, 8) |
| assert forcer.generator_ids == [id(generator), id(generator)] |
| assert torch.equal( |
| torch.cat((result.first.token_rows, result.second.token_rows)), |
| reference, |
| ) |
| assert result.first.cache.sequence_length == 129 |
| assert result.second.resume_from_sequence_length == 129 |
| assert result.second.resume_entry_sequence_length == 130 |
| assert len({ |
| result.generator_initial_state_sha256, |
| result.generator_after_first_sha256, |
| result.generator_after_second_sha256, |
| }) == 3 |
|
|
|
|
| def test_config_rejects_identity_or_scope_drift(tmp_path: Path): |
| source = (ROOT / "configs/continuation-tier-a.yaml").read_text() |
| changed = tmp_path / "changed.yaml" |
| changed.write_text(source.replace(BASE_ID, "f" * 64)) |
| with pytest.raises(ValueError, match="identity"): |
| load_continuation_config(changed) |
|
|
|
|
| def test_report_template_keeps_negative_claims(): |
| text = (ROOT / "reports/CONTINUATION.md").read_text() |
| assert "NOT_RUN" in text |
| assert "Arbitrary-WAV continuation remains BLOCKED" in text |
|
|
