File size: 4,879 Bytes
90884df | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 | """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
|