File size: 11,560 Bytes
ae36cfb | 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 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 | """A100 opt-in protocol/budget checks; no GPU, model, network, or engine execution."""
import copy
import hashlib
import json
import sys
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts'))
import benchmark_batch
import evaluate_pro as pro
class ProEvaluationTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.root = Path(self.temp.name)
self.run = self.root / 'run'
self.data = self.root / 'data'
self.adapter = self.run / 'adapter'
self.output = self.run / pro.NAME
self.t4 = self.run / 'evaluation'
for path in (self.data, self.adapter, self.t4):
path.mkdir(parents=True)
rows = [{'instruction': f'original training prompt {index}', 'output': '{}'} for index in range(4)]
(self.data / 'train.jsonl').write_text(''.join(json.dumps(row) + '\n' for row in rows))
for name in ('dev.jsonl', 'eval_prompts.jsonl', 'validation_manifest.jsonl'):
(self.data / name).write_text('original fixture ' + name)
signature = {'base_model': pro.BASE_MODEL, 'base_revision': 'pinned-revision',
'system_prompt_sha256': hashlib.sha256(pro.SYSTEM_PROMPT.encode()).hexdigest()}
for key, filename in [('train_sha256', 'train.jsonl'), ('dev_sha256', 'dev.jsonl'),
('blind_sha256', 'eval_prompts.jsonl'), ('validation_manifest_sha256', 'validation_manifest.jsonl')]:
signature[key] = pro.file_hash(self.data / filename)
for path in (self.run / 'run_manifest.json', self.adapter / 'run_manifest.json'):
path.write_text(json.dumps({'signature': signature}))
(self.adapter / 'TRAINING_COMPLETE.json').write_text(json.dumps({'status': 'complete', 'target_steps': 14, 'global_step': 14}))
(self.adapter / 'adapter_model.safetensors').write_bytes(b'fixture-adapter-no-real-tensors')
(self.adapter / 'tokenizer.json').write_text('{}')
self.godot = self.root / 'godot-not-an-executable'
self.godot.write_bytes(b'fixed-engine-fixture')
self.runtime = {'cuda_available': True, 'device_count': 1, 'gpu': 'NVIDIA A100-SXM4-40GB',
'torch_version': '2.11.0+cu130', 'cuda_version': '13.0',
'packages': {'unsloth': 'fixture', 'transformers': 'fixture', 'peft': 'fixture', 'bitsandbytes': 'fixture'}}
self.prior = {'backend': 'unsloth', 'adapter': pro.directory_fingerprint(self.adapter),
'base_model': signature['base_model'], 'base_revision': signature['base_revision'],
'system_prompt_sha256': signature['system_prompt_sha256'],
'gpu': 'Tesla T4', 'base_artifacts': {'sha256': 'fixture-pinned-base', 'files': {}},
**{key: self.runtime[key] for key in ('torch_version', 'cuda_version', 'packages')}}
self.prior_manifest = {'expected_count': 500, 'max_new_tokens': 3072, 'run_seconds': 60,
'sample_indices': [0, 200, 400], 'smoke_test': False, 'batch_size': 8,
'workers': 4, 'decoding': {'temperature': 0, 'seed': 42}, 'latency_version': 2,
'prompts_sha256': signature['blind_sha256'], 'runtime_sha256': pro.file_hash(self.godot),
'generation': self.prior}
(self.t4 / 'manifest.json').write_text(json.dumps(self.prior_manifest))
(self.t4 / 'result-fixture.json').write_text('immutable prior result')
self.prompts = [row['instruction'] for row in rows]
def tearDown(self):
self.temp.cleanup()
def protocol(self):
return pro.validate_sources(self.run, self.data, self.output, self.godot, self.runtime)
def capacity(self):
protocol = self.protocol()
estimate = {'within_limit': True, 'assessed_batch_size': 8, 'target_context_tokens': 4096,
'cache_growth_allowance_tokens': 512, 'reserve_bytes': 2_000_000_000,
'estimated_peak_bytes': 10_000_000_000, 'effective_limit_bytes': 12_500_000_000}
full = {'status': 'success', 'batch_size': 8, 'max_new_tokens': 3072, 'within_memory_limit': True,
'full_context_memory_estimate': estimate, 'peak_allocated_bytes': 8_000_000_000,
'peak_reserved_bytes': 9_000_000_000, 'per_prompt_tokens': [1536] * 8,
'end_to_end_call_seconds': 20.0}
return {'status': 'complete', 'evaluation_profile': 'pro-a100-opt-in', 'blind_prompts_read': False,
'batch_sizes': [8], 'recommended_batch_size': 8, 'full_max_new_tokens': 3072,
'memory_estimate_policy': 'kv_full_context_v1', 'training_jsonl_sha256': protocol['train_sha256'],
'training_prompts_sha256': hashlib.sha256(json.dumps(self.prompts, ensure_ascii=False).encode()).hexdigest(),
'gpu': self.runtime['gpu'], 'gpu_memory_bytes': 40_000_000_000,
'generation': {**self.prior, 'gpu': self.runtime['gpu']},
'full_check': full, 'measurements': [{**full, 'max_new_tokens': 256}]}
def test_valid_predeclaration_is_immutable_and_t4_evidence_untouched(self):
original = {path.name: path.read_bytes() for path in self.t4.iterdir()}
protocol = self.protocol()
pro.declare_protocol(self.output, protocol)
first = (self.output / 'pro-protocol.json').read_bytes()
pro.declare_protocol(self.output, protocol)
self.assertEqual(first, (self.output / 'pro-protocol.json').read_bytes())
self.assertEqual(original, {path.name: path.read_bytes() for path in self.t4.iterdir()})
self.assertFalse((self.output / 'items').exists())
with self.assertRaisesRegex(ValueError, 'protocol changed'):
pro.declare_protocol(self.output, {**protocol, 'batch_size': 16})
self.assertEqual(first, (self.output / 'pro-protocol.json').read_bytes())
def test_t4_and_other_output_paths_and_symlink_are_rejected(self):
for output in (self.t4, self.run / 'evaluation-a100-v2', self.root / pro.NAME):
with self.subTest(output=output), self.assertRaisesRegex(ValueError, 'separate'):
pro.validate_sources(self.run, self.data, output, self.godot, self.runtime)
self.output.symlink_to(self.t4, target_is_directory=True)
with self.assertRaisesRegex(ValueError, 'separate'):
self.protocol()
def test_no_unowned_results_are_adopted(self):
self.output.mkdir()
(self.output / 'manifest.json').write_text('previously started')
with self.assertRaisesRegex(ValueError, 'without the A100 predeclaration'):
pro.declare_protocol(self.output, self.protocol())
def test_actual_a100_and_unchanged_runtime_are_required(self):
for changes in ({'gpu': 'Tesla T4'}, {'cuda_available': False}, {'device_count': 2},
{'torch_version': 'changed'}, {'packages': {}}):
with self.subTest(changes=changes), self.assertRaises(ValueError):
pro.validate_sources(self.run, self.data, self.output, self.godot, {**self.runtime, **changes})
def test_changed_tokenizer_data_or_incomplete_training_is_rejected(self):
for path, content in [(self.adapter / 'tokenizer.json', '{"changed":true}'),
(self.data / 'train.jsonl', 'changed'),
(self.adapter / 'TRAINING_COMPLETE.json', '{"status":"partial"}')]:
old = path.read_bytes()
path.write_text(content)
with self.subTest(path=path), self.assertRaises(ValueError):
self.protocol()
path.write_bytes(old)
def test_capacity_accepts_only_matching_completed_training_probe(self):
report = self.capacity()
self.assertTrue(pro.capacity_valid(report, self.protocol(), self.prior, self.prompts))
for change in ({'status': 'full_length_check'}, {'recommended_batch_size': 16}, {'blind_prompts_read': True},
{'evaluation_profile': 't4'}, {'training_jsonl_sha256': 'other'}, {'gpu': 'Tesla T4'}):
with self.subTest(change=change):
self.assertFalse(pro.capacity_valid({**report, **change}, self.protocol(), self.prior, self.prompts))
changed = copy.deepcopy(report)
changed['full_check']['full_context_memory_estimate']['within_limit'] = False
self.assertFalse(pro.capacity_valid(changed, self.protocol(), self.prior, self.prompts))
changed = copy.deepcopy(report)
changed['generation']['base_revision'] = 'different-base'
self.assertFalse(pro.capacity_valid(changed, self.protocol(), self.prior, self.prompts))
def test_budget_normalizes_early_eos_and_never_credits_overlap(self):
budget = pro.budget_projection(self.capacity()['full_check'], 10, 100, 360)
self.assertEqual(budget['normalized_3072_batch_seconds'], 40)
self.assertEqual(budget['generation_seconds_with_margin'], 63 * 40 * 1.5)
self.assertEqual(budget['validation_seconds_without_overlap_credit'], 125 * 90)
self.assertEqual(budget['projected_remaining_seconds'], 63 * 40 * 1.5 + 125 * 90 + 1800)
self.assertEqual(budget['estimated_units_used_since_start'], 1)
self.assertEqual(budget['usable_remaining_units'], 89)
self.assertTrue(budget['fits_observed_balance'])
self.assertFalse(pro.budget_projection(self.capacity()['full_check'], 10, 40, 360)['fits_observed_balance'])
self.assertFalse(pro.budget_projection(self.capacity()['full_check'], 10, 100, 40000)['fits_observed_balance'])
def test_invalid_budget_observations_or_measurements_fail_closed(self):
for rate, balance in [(0, 100), (10, 0), (float('nan'), 100), (10, float('inf')), (-1, 100)]:
with self.subTest(rate=rate, balance=balance), self.assertRaises(ValueError):
pro.budget_projection(self.capacity()['full_check'], rate, balance, 0)
for changed in ({'per_prompt_tokens': [0] * 8}, {'end_to_end_call_seconds': float('nan')}):
with self.subTest(changed=changed), self.assertRaises(ValueError):
pro.budget_projection({**self.capacity()['full_check'], **changed}, 10, 100, 0)
def test_opt_in_capacity_preserves_t4_hardware_default(self):
def fake(name):
return SimpleNamespace(cuda=SimpleNamespace(is_available=lambda: True, device_count=lambda: 1,
get_device_name=lambda: name))
benchmark_batch.require_gpu(fake('Tesla T4'))
with self.assertRaises(RuntimeError):
benchmark_batch.require_gpu(fake('NVIDIA A100-SXM4-40GB'))
benchmark_batch.require_gpu(fake('NVIDIA A100-SXM4-40GB'), pro_a100=True)
with self.assertRaises(RuntimeError):
benchmark_batch.require_gpu(fake('Tesla T4'), pro_a100=True)
def test_capacity_reads_only_hash_matched_training_split(self):
rows = benchmark_batch.capacity_prompts(self.adapter, self.data / 'train.jsonl')
self.assertEqual([row['instruction'] for row in rows], self.prompts)
with self.assertRaisesRegex(ValueError, 'training split'):
benchmark_batch.capacity_prompts(self.adapter, self.data / 'eval_prompts.jsonl')
if __name__ == '__main__':
unittest.main()
|