| |
| """Opt in to a separate, budget-checked A100 evaluation; never alter the T4 run.""" |
| from __future__ import annotations |
|
|
| import argparse |
| import hashlib |
| import json |
| import math |
| import os |
| import signal |
| import subprocess |
| import sys |
| import time |
| from datetime import datetime, timezone |
| from pathlib import Path |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| sys.path.insert(0, str(ROOT / 'src')) |
| sys.path.insert(0, str(ROOT / 'scripts')) |
| from generate import directory_fingerprint, file_hash |
|
|
| from vibegame import BASE_MODEL, SYSTEM_PROMPT |
|
|
| NAME = 'evaluation-a100-v1' |
| BATCH = 8 |
| MAX_TOKENS = 3072 |
| COUNT = 500 |
| WORKERS = 4 |
| GENERATION_MARGIN = 1.5 |
| VALIDATION_SECONDS_PER_PROMPT = 90 |
| SETUP_SECONDS = 1800 |
| UNIT_RESERVE_FRACTION = 0.1 |
| RATIONALE = ('Predeclared separate full A100 pass to reduce completion time after the free T4 quota was exhausted. ' |
| 'The completed adapter and decoding are unchanged. The A100 500-prompt pass is the primary final ' |
| 'protocol; the partial T4 run remains separate and will be disclosed. No prompt-level results ' |
| 'are pooled or selected between runs. Capacity decisions use only admitted training prompts.') |
|
|
|
|
| def read_json(path): |
| return json.loads(path.read_text()) |
|
|
|
|
| def write_new(path, value): |
| with path.open('x') as handle: |
| json.dump(value, handle, sort_keys=True, indent=2) |
| handle.write('\n') |
| handle.flush() |
| os.fsync(handle.fileno()) |
|
|
|
|
| def next_path(output, prefix): |
| for number in range(10000): |
| path = output / f'{prefix}-{number:04d}.json' |
| if not path.exists(): |
| return path |
| raise RuntimeError('Too many preserved attempts; inspect this evaluation directory') |
|
|
|
|
| def runtime_snapshot(): |
| |
| probe = '''import torch,importlib.metadata as m,json |
| names=['unsloth','transformers','peft','bitsandbytes'] |
| print(json.dumps({'cuda_available':torch.cuda.is_available(),'device_count':torch.cuda.device_count(), |
| 'gpu':torch.cuda.get_device_name(0) if torch.cuda.is_available() else None, |
| 'torch_version':torch.__version__,'cuda_version':torch.version.cuda, |
| 'packages':{name:m.version(name) for name in names}}))''' |
| return json.loads(subprocess.check_output([sys.executable, '-c', probe], text=True)) |
|
|
|
|
| def validate_sources(run, data, output, godot, runtime): |
| if output.is_symlink() or output.resolve() != run.resolve() / NAME: |
| raise ValueError('A100 evaluation must use the separate RUN/evaluation-a100-v1 directory') |
| if not runtime.get('cuda_available') or runtime.get('device_count') != 1 or 'A100' not in str(runtime.get('gpu')): |
| raise ValueError('This entry point requires one already allocated A100 CUDA GPU') |
| adapter = run / 'adapter' |
| completed = read_json(adapter / 'TRAINING_COMPLETE.json') |
| if (completed.get('status') != 'complete' or completed.get('target_steps', 0) <= 0 |
| or completed.get('global_step') != completed.get('target_steps')): |
| raise ValueError('A100 evaluation requires the completed training adapter') |
| saved = read_json(adapter / 'run_manifest.json')['signature'] |
| if read_json(run / 'run_manifest.json')['signature'] != saved: |
| raise ValueError('Adapter and original training manifest disagree') |
| for key, filename in [('train_sha256', 'train.jsonl'), ('dev_sha256', 'dev.jsonl'), |
| ('blind_sha256', 'eval_prompts.jsonl'), ('validation_manifest_sha256', 'validation_manifest.jsonl')]: |
| if file_hash(data / filename) != saved.get(key): |
| raise ValueError('Original frozen data changed: ' + filename) |
| system_hash = hashlib.sha256(SYSTEM_PROMPT.encode()).hexdigest() |
| if saved.get('base_model') != BASE_MODEL or saved.get('system_prompt_sha256') != system_hash: |
| raise ValueError('Model or system prompt differs from the completed training run') |
| prior_path = run / 'evaluation/manifest.json' |
| prior = read_json(prior_path) |
| generation = prior['generation'] |
| if (prior.get('expected_count') != COUNT or prior.get('max_new_tokens') != MAX_TOKENS |
| or prior.get('run_seconds') != 60 or prior.get('sample_indices') != [0, 200, 400] |
| or prior.get('smoke_test') is not False or prior.get('batch_size') != BATCH |
| or prior.get('workers') != WORKERS or prior.get('decoding') != {'temperature': 0, 'seed': 42} |
| or prior.get('latency_version') != 2 or 'T4' not in generation.get('gpu', '') |
| or prior.get('prompts_sha256') != saved['blind_sha256'] |
| or prior.get('runtime_sha256') != file_hash(godot)): |
| raise ValueError('The original T4 evaluation does not match the fixed protocol') |
| if (generation.get('backend') != 'unsloth' or generation.get('base_model') != saved['base_model'] |
| or generation.get('base_revision') != saved['base_revision'] |
| or generation.get('system_prompt_sha256') != system_hash |
| or generation.get('adapter') != directory_fingerprint(adapter)): |
| raise ValueError('Completed adapter/tokenizer identity differs from the original T4 run') |
| for key in ('torch_version', 'cuda_version', 'packages'): |
| if runtime.get(key) != generation.get(key): |
| raise ValueError('Restore the original evaluation runtime before A100 evaluation: ' + key) |
| return {'schema_version': 1, 'experiment': NAME, 'rationale': RATIONALE, |
| 'hardware_tier': 'Colab Pro A100; training previously completed on a free T4', |
| 'source_t4_manifest_sha256': file_hash(prior_path), 'source_t4_directory': str(prior_path.parent.resolve()), |
| 'adapter': generation['adapter'], 'base_model': saved['base_model'], 'base_revision': saved['base_revision'], |
| 'system_prompt_sha256': system_hash, 'train_sha256': saved['train_sha256'], |
| 'prompts_sha256': saved['blind_sha256'], 'godot_sha256': prior['runtime_sha256'], |
| 'gpu_runtime': runtime, 'expected_count': COUNT, 'batch_size': BATCH, 'workers': WORKERS, |
| 'max_new_tokens': MAX_TOKENS, 'run_seconds': 60, 'sample_indices': [0, 200, 400], |
| 'decoding': prior['decoding'], 'dtype': 'float16', 'base_quantization': 'unchanged pinned 4-bit base', |
| 't4_evidence_policy': 'Read only; preserve every result and interrupted attempt; never pool scores', |
| 'source_sha256': {name: file_hash(ROOT / 'scripts' / name) for name in |
| ('generate.py', 'evaluate.py', 'benchmark_batch.py', 'evaluate_pro.py')}, |
| 'budget_policy': {'generation_margin': GENERATION_MARGIN, 'validation_seconds_per_prompt': VALIDATION_SECONDS_PER_PROMPT, |
| 'setup_seconds': SETUP_SECONDS, 'balance_reserve_fraction': UNIT_RESERVE_FRACTION, |
| 'overlap_credit': False, |
| 'limitations': 'Conservative estimate, not a guaranteed upper bound or a billing cap; throughput, import time and Colab rates can vary.'}} |
|
|
|
|
| def declare_protocol(output, protocol): |
| path = output / 'pro-protocol.json' |
| if path.exists(): |
| if read_json(path).get('protocol') != protocol: |
| raise ValueError('A100 protocol changed; original predeclaration and evidence are preserved') |
| else: |
| if output.exists() and any(output.iterdir()): |
| raise ValueError('Refusing to adopt existing results without the A100 predeclaration') |
| output.mkdir(parents=True, exist_ok=True) |
| write_new(path, {'declared_utc': datetime.now(timezone.utc).isoformat(), 'protocol': protocol}) |
|
|
|
|
| def capacity_valid(report, protocol, prior_generation, training_prompts): |
| expected_prompt_hash = hashlib.sha256(json.dumps(training_prompts, ensure_ascii=False).encode()).hexdigest() |
| generation = report.get('generation', {}) |
| if (report.get('status') != 'complete' or report.get('evaluation_profile') != 'pro-a100-opt-in' |
| or report.get('blind_prompts_read') is not False or report.get('batch_sizes') != [BATCH] |
| or report.get('recommended_batch_size') != BATCH or report.get('full_max_new_tokens') != MAX_TOKENS |
| or report.get('memory_estimate_policy') != 'kv_full_context_v1' |
| or report.get('training_jsonl_sha256') != protocol['train_sha256'] |
| or report.get('training_prompts_sha256') != expected_prompt_hash |
| or report.get('gpu') != protocol['gpu_runtime']['gpu']): |
| return False |
| for key in ('backend', 'adapter', 'base_model', 'base_revision', 'torch_version', 'cuda_version', 'packages', |
| 'system_prompt_sha256', 'base_artifacts'): |
| if key not in prior_generation or generation.get(key) != prior_generation[key]: |
| return False |
| if generation.get('gpu') != protocol['gpu_runtime']['gpu']: |
| return False |
| full = report.get('full_check') or {} |
| short = next((row for row in report.get('measurements', []) if row.get('batch_size') == BATCH), {}) |
| for row in (short, full): |
| estimate = row.get('full_context_memory_estimate') or {} |
| if (row.get('status') != 'success' or row.get('batch_size') != BATCH |
| or estimate.get('within_limit') is not True or estimate.get('assessed_batch_size') != BATCH |
| or estimate.get('target_context_tokens') != 4096 or estimate.get('cache_growth_allowance_tokens', 0) < 512 |
| or estimate.get('reserve_bytes', 0) < 2_000_000_000 |
| or not 0 < estimate.get('estimated_peak_bytes', 0) < estimate.get('effective_limit_bytes', 0) |
| or estimate.get('effective_limit_bytes', 0) > min(12_500_000_000, report.get('gpu_memory_bytes', 0) - 2_000_000_000) |
| or not 0 < row.get('peak_allocated_bytes', 0) < 12_500_000_000 |
| or not 0 < row.get('peak_reserved_bytes', 0) < 12_500_000_000): |
| return False |
| return full.get('max_new_tokens') == MAX_TOKENS and full.get('within_memory_limit') is True |
|
|
|
|
| def budget_projection(full, rate, available, elapsed, fresh_prompts=COUNT, pending_results=COUNT): |
| if any(not math.isfinite(x) or x <= 0 for x in (rate, available)) or not math.isfinite(elapsed) or elapsed < 0: |
| raise ValueError('Observed compute-unit rate and available balance must be finite and positive') |
| tokens = full.get('per_prompt_tokens', []) |
| seconds = full.get('end_to_end_call_seconds') |
| if (len(tokens) != BATCH or any(type(x) is not int or not 0 < x <= MAX_TOKENS for x in tokens) |
| or not isinstance(seconds, (int, float)) or not math.isfinite(seconds) or seconds <= 0): |
| raise ValueError('Budget requires a valid measured full-cap batch') |
| if not 0 <= fresh_prompts <= COUNT or not 0 <= pending_results <= COUNT: |
| raise ValueError('Invalid remaining evaluation count') |
| normalized = seconds * MAX_TOKENS / max(tokens) |
| generation = math.ceil(fresh_prompts / BATCH) * normalized * GENERATION_MARGIN |
| validation = math.ceil(pending_results / WORKERS) * VALIDATION_SECONDS_PER_PROMPT |
| projected = generation + validation + SETUP_SECONDS |
| spent = elapsed * rate / 3600 |
| reserve = available * UNIT_RESERVE_FRACTION |
| usable = available - spent - reserve |
| units = projected * rate / 3600 |
| return {'observed_compute_units_per_hour': rate, 'observed_available_units_at_start': available, |
| 'elapsed_session_seconds': elapsed, 'estimated_units_used_since_start': spent, 'reserved_units': reserve, |
| 'fresh_prompts': fresh_prompts, 'pending_results': pending_results, |
| 'measured_batch_seconds': seconds, 'normalized_3072_batch_seconds': normalized, |
| 'generation_seconds_with_margin': generation, 'validation_seconds_without_overlap_credit': validation, |
| 'setup_seconds': SETUP_SECONDS, 'projected_remaining_seconds': projected, |
| 'projected_remaining_units': units, 'usable_remaining_units': usable, |
| 'fits_observed_balance': units <= usable, |
| 'budget_deadline_remaining_seconds': max(0, usable * 3600 / rate)} |
|
|
|
|
| def run_budgeted(command, seconds): |
| if seconds <= 0: |
| raise RuntimeError('No observed compute-unit budget remains') |
| process = subprocess.Popen([str(x) for x in command], cwd=ROOT, start_new_session=True) |
| try: |
| code = process.wait(timeout=seconds) |
| except subprocess.TimeoutExpired: |
| os.killpg(process.pid, signal.SIGTERM) |
| try: |
| process.wait(timeout=10) |
| except subprocess.TimeoutExpired: |
| os.killpg(process.pid, signal.SIGKILL) |
| process.wait() |
| raise RuntimeError('Stopped at the observed compute-unit time budget; immutable evidence was retained') from None |
| if code: |
| raise subprocess.CalledProcessError(code, command) |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(description=__doc__) |
| for name in ('run', 'data', 'godot'): |
| parser.add_argument('--' + name, type=Path, required=True) |
| parser.add_argument('--output', type=Path) |
| parser.add_argument('--compute-units-per-hour', type=float, required=True, help='Current A100 rate observed in Colab') |
| parser.add_argument('--available-compute-units', type=float, required=True, help='Current remaining balance observed immediately before this invocation') |
| args = parser.parse_args() |
| started = time.monotonic() |
| rate, balance = args.compute_units_per_hour, args.available_compute_units |
| if any(not math.isfinite(x) or x <= 0 for x in (rate, balance)): |
| parser.error('Both observed compute-unit values must be finite and positive') |
| args.run, args.data, args.godot = (path.expanduser().resolve(strict=True) for path in (args.run, args.data, args.godot)) |
| drive = Path('/content/drive/MyDrive') |
| if not drive.is_dir() or not args.run.is_relative_to(drive.resolve()) or not args.data.is_relative_to(drive.resolve()): |
| parser.error('Run, original data and A100 evidence must persist on mounted Drive') |
| output = args.output or args.run / NAME |
| runtime = runtime_snapshot() |
| protocol = validate_sources(args.run, args.data, output, args.godot, runtime) |
| import fcntl |
| lock_name = hashlib.sha256(str(output.resolve()).encode()).hexdigest() |
| with (Path('/content') / f'.vibegame-pro-{lock_name}.lock').open('a+') as lock: |
| fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) |
| declare_protocol(output, protocol) |
| prior_generation = read_json(args.run / 'evaluation/manifest.json')['generation'] |
| from benchmark_batch import capacity_prompts |
| training_prompts = [row['instruction'] for row in capacity_prompts(args.run / 'adapter', args.data / 'train.jsonl')] |
| capacity_path, capacity = None, None |
| for path in sorted(output.glob('capacity-*.json')): |
| try: |
| candidate = read_json(path) |
| except (ValueError, OSError): |
| continue |
| if capacity_valid(candidate, protocol, prior_generation, training_prompts): |
| capacity_path, capacity = path, candidate |
| break |
| if capacity is None: |
| capacity_path = next_path(output, 'capacity') |
| remaining = balance * (1 - UNIT_RESERVE_FRACTION) * 3600 / rate - (time.monotonic() - started) |
| run_budgeted([sys.executable, ROOT / 'scripts/benchmark_batch.py', capacity_path, |
| '--adapter', args.run / 'adapter', '--batch-sizes', BATCH, '--pro-a100', |
| '--training-jsonl', args.data / 'train.jsonl'], remaining) |
| capacity = read_json(capacity_path) |
| if not capacity_valid(capacity, protocol, prior_generation, training_prompts): |
| raise RuntimeError('A100 batch8 capacity or exact model/runtime identity check failed; no blind attempts started') |
| items = [output / 'items' / f'{index:04d}' for index in range(COUNT)] |
| fresh = sum(not (item / 'attempt.json').exists() for item in items) |
| pending = sum(not (item / 'result.json').exists() for item in items) |
| projection = budget_projection(capacity['full_check'], rate, balance, time.monotonic() - started, fresh, pending) |
| budget_path = next_path(output, 'budget') |
| write_new(budget_path, {'observed_utc': datetime.now(timezone.utc).isoformat(), |
| 'capacity_report_sha256': file_hash(capacity_path), **projection}) |
| if not projection['fits_observed_balance']: |
| raise RuntimeError('Conservative A100 evaluation projection exceeds the observed balance; no blind attempts started. See ' + str(budget_path)) |
| command = [sys.executable, ROOT / 'scripts/evaluate.py', '--prompts', args.data / 'eval_prompts.jsonl', |
| '--output', output, '--backend', 'unsloth', '--adapter', args.run / 'adapter', |
| '--godot', args.godot, '--run-seconds', 60, '--expected-count', COUNT, |
| '--max-new-tokens', MAX_TOKENS, '--batch-size', BATCH, '--workers', WORKERS] |
| if prior_generation.get('requested_base_model') is not None: |
| command += ['--base-model', prior_generation['requested_base_model']] |
| run_budgeted(command, projection['budget_deadline_remaining_seconds']) |
|
|
|
|
| if __name__ == '__main__': |
| main() |
|
|
