"""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()