File size: 5,667 Bytes
312fb3f | 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 | from pathlib import Path
import pytest
import tempfile
from recurrent_sampler_repro.evidence import audit_source_ast, SAMPLER_GIT_BLOB
def get_project_root() -> Path:
return Path(__file__).resolve().parent.parent
def test_source_ast_audit_derived_operation_order():
project_root = get_project_root()
res = audit_source_ast(project_root)
assert res["git_blob"] == SAMPLER_GIT_BLOB
assert res["dispatcher"]["function"] == "generate"
assert res["dispatcher"]["dispatches_to_diffusion_style"] is True
sampler_info = res["sampler"]
assert sampler_info["function"] == "generate_diffusion_style"
assert sampler_info["defaults"]["headway"] == 1
assert sampler_info["defaults"]["inner_recurrence"] == 4
assert sampler_info["defaults"]["freeze_strategy"] == "latent-diff"
assert sampler_info["defaults"]["max_wavefront"] == 128
ctrl = sampler_info["control_flow"]
assert ctrl["operation_order_valid"] is True
assert ctrl["operations"] == [
"recurrent_iterate",
"prediction_logits",
"sampling",
"state_append",
"prefix_max_wavefront_truncation",
"latent_diff_freezing",
]
assert ctrl["latent_diff_normalized_predicate_found"] is True
def test_source_ast_audit_mutation_missing_iterate_fails():
project_root = get_project_root()
sampler_bytes = (
project_root / "vendor" / "recurrent-pretraining" / "recpre" / "raven_modeling_minimal.py"
).read_text(encoding="utf-8")
mutated_bytes = sampler_bytes.replace("self.iterate_one_step", "self._disabled_iterate")
with tempfile.TemporaryDirectory() as tmpdir:
tmp_path = Path(tmpdir)
(tmp_path / "vendor" / "recurrent-pretraining" / "recpre").mkdir(parents=True)
(tmp_path / "vendor" / "recurrent-pretraining" / "recpre" / "raven_modeling_minimal.py").write_text(
mutated_bytes, encoding="utf-8"
)
with pytest.raises(ValueError, match="Missing or reordered sampler operation"):
audit_source_ast(tmp_path)
def test_source_ast_audit_mutation_unrelated_norm_fails():
project_root = get_project_root()
sampler_bytes = (
project_root / "vendor" / "recurrent-pretraining" / "recpre" / "raven_modeling_minimal.py"
).read_text(encoding="utf-8")
# Replace normalized latent diff with unnormalized norm inside the branch
old_line = 'criterion = (match_states - matching_prev_states).norm(dim=-1) / match_states.norm(dim=-1)'
new_line = 'criterion = (match_states - matching_prev_states).norm(dim=-1)'
mutated_bytes = sampler_bytes.replace(old_line, new_line)
with tempfile.TemporaryDirectory() as tmpdir:
tmp_path = Path(tmpdir)
(tmp_path / "vendor" / "recurrent-pretraining" / "recpre").mkdir(parents=True)
(tmp_path / "vendor" / "recurrent-pretraining" / "recpre" / "raven_modeling_minimal.py").write_text(
mutated_bytes, encoding="utf-8"
)
with pytest.raises(ValueError, match="Normalized latent-difference freezing predicate not found"):
audit_source_ast(tmp_path)
def test_source_ast_audit_matching_string_elsewhere_does_not_pass():
project_root = get_project_root()
sampler_bytes = (
project_root / "vendor" / "recurrent-pretraining" / "recpre" / "raven_modeling_minimal.py"
).read_text(encoding="utf-8")
# Insert matching string as comment outside branch, but break actual criterion assignment inside branch
old_line = 'criterion = (match_states - matching_prev_states).norm(dim=-1) / match_states.norm(dim=-1)'
comment_line = ' # (match_states - matching_prev_states).norm(dim=-1) / match_states.norm(dim=-1)'
broken_line = 'criterion = (match_states - matching_prev_states).norm(dim=-1)'
mutated_bytes = sampler_bytes.replace(old_line, f'{comment_line}\n {broken_line}')
with tempfile.TemporaryDirectory() as tmpdir:
tmp_path = Path(tmpdir)
(tmp_path / "vendor" / "recurrent-pretraining" / "recpre").mkdir(parents=True)
(tmp_path / "vendor" / "recurrent-pretraining" / "recpre" / "raven_modeling_minimal.py").write_text(
mutated_bytes, encoding="utf-8"
)
with pytest.raises(ValueError, match="Normalized latent-difference freezing predicate not found"):
audit_source_ast(tmp_path)
def test_source_ast_audit_mutation_false_sibling_criterion_fails():
project_root = get_project_root()
sampler_bytes = (
project_root / "vendor" / "recurrent-pretraining" / "recpre" / "raven_modeling_minimal.py"
).read_text(encoding="utf-8")
# Break real criterion and insert if False block containing criterion assignment inside outer latent check
old_line = 'criterion = (match_states - matching_prev_states).norm(dim=-1) / match_states.norm(dim=-1)'
broken_line = 'criterion = (match_states - matching_prev_states).norm(dim=-1)'
false_sibling = 'if False:\n criterion = (match_states - matching_prev_states).norm(dim=-1) / match_states.norm(dim=-1)'
mutated_bytes = sampler_bytes.replace(old_line, f'{broken_line}\n {false_sibling}')
with tempfile.TemporaryDirectory() as tmpdir:
tmp_path = Path(tmpdir)
(tmp_path / "vendor" / "recurrent-pretraining" / "recpre").mkdir(parents=True)
(tmp_path / "vendor" / "recurrent-pretraining" / "recpre" / "raven_modeling_minimal.py").write_text(
mutated_bytes, encoding="utf-8"
)
with pytest.raises(ValueError, match="Normalized latent-difference freezing predicate not found"):
audit_source_ast(tmp_path)
|