| |
| """Prepare or explicitly launch the isolated 80/15/5 continuation.""" |
| import argparse |
| import copy |
| from datetime import datetime |
| import gzip |
| import json |
| from pathlib import Path |
| import shutil |
| import sys |
| import uuid |
|
|
| ROOT=Path(__file__).resolve().parents[1] |
| PREPARED=ROOT/'prepared/python_sql_20261104' |
| sys.path.insert(0,str(PREPARED/'runtime')) |
| from sml_v1.common import atomic_json,file_sha256,fingerprint,load_corpus,lock,read_json |
| from sml_v1.code_phase import KEY,phase_recipe,training_manifest,migrate_cursor,CodePhaseStream |
| from sml_v1.code_validation import load_panel |
| from sml_v1.continuation import check_resume |
| from sml_v1.recipe import tokens_per_update,validate |
| from sml_v1.stable_schedule import effective_learning_rate |
| from sml_v1.lr_scheduler import restore_scheduler |
| from sml_v1.tokenization import Tokenizer |
|
|
| PARENT=ROOT/'runs/deep44_prose_gradual_100b_20261028' |
| OUTPUT=ROOT/'runs/deep44_prose_python_sql_80_15_5_20261104' |
| NETWORK=ROOT/'diagnostics/m5_migration_20260922/network_20261102_234513' |
| TOKENIZER=ROOT.parent/'reference/opensml-150m/tokenizer/bytebpe32k_v1' |
|
|
|
|
| def checkpoint_path(pointer): |
| if pointer.is_symlink():raise ValueError('Checkpoint pointer cannot be a symlink') |
| info=read_json(pointer);name=info.get('bundle','') |
| if info.get('format')!='sml-pretrain-bundle-v1' or not name.startswith('step_') or Path(name).name!=name: |
| raise ValueError('Invalid full checkpoint pointer') |
| source=pointer.parent/name |
| if source.is_symlink():raise ValueError('Checkpoint cannot be a symlink') |
| return source |
|
|
|
|
| def modules(): |
| return {p.name:file_sha256(p) for p in (PREPARED/'runtime/sml_v1').glob('*.py')} |
|
|
|
|
| def verify_readiness(): |
| readiness=read_json(PREPARED/'readiness.json') |
| if readiness.get('status')!='ready-for-idle-cluster-check': |
| raise ValueError('Prepared continuation has not passed its CPU/data checks') |
| for name,digest in readiness['files'].items(): |
| path=ROOT.parent/name |
| if not path.is_file() or file_sha256(path)!=digest: |
| raise ValueError('Prepared input changed after validation: '+name) |
| from sml_v1.code_sources import parser |
| parser() |
| return readiness |
|
|
|
|
| def plan(*,full_verify=False): |
| readiness=verify_readiness() |
| resume=(OUTPUT/'latest.json').exists() |
| source=checkpoint_path((OUTPUT if resume else PARENT)/'latest.json') |
| if full_verify: |
| from sml_v1.checkpoint_metadata import verify_bundle_metadata |
| metadata=verify_bundle_metadata(source) |
| else: |
| metadata=read_json(source/'model.safetensors.json') |
| tokenizer=Tokenizer(TOKENIZER) |
| corpus=load_corpus(PREPARED/'corpus.json') |
| from sml_v1.sample_prompts import read_prompts |
| read_prompts(PREPARED/'sample_prompts.json') |
| manifest=training_manifest(corpus,tokenizer,metadata['recipe']['stream']) |
| recipe=validate(copy.deepcopy(metadata['recipe'])) if resume else phase_recipe(metadata,manifest) |
| if KEY not in recipe or recipe[KEY]['manifest']!=fingerprint(manifest): |
| raise ValueError('Existing phase must resume its own unchanged data contract') |
| if tokenizer.vocab_size!=recipe['model']['vocab_size'] or len(recipe['batches'])!=5: |
| raise ValueError('Model/tokenizer or five-rank configuration differs') |
| contract=dict(recipe=fingerprint(recipe),tokenizer=tokenizer.fingerprint, |
| data=fingerprint(manifest),code=fingerprint(modules())) |
| transition=check_resume(metadata,contract,recipe,manifest) |
| load_panel(PREPARED/'code_validation.npz',manifest['code_validation'],manifest,tokenizer,recipe) |
| if full_verify: |
| with gzip.open(source/'model.safetensors.rank0.data_state.json.gz','rt') as f: |
| saved=json.load(f)['stream_state'] |
| cursor=saved if resume else migrate_cursor(saved,metadata,manifest,recipe) |
| check=CodePhaseStream(manifest,tokenizer,'train',recipe['model']['max_seq_len'],cursor, |
| exclusions=manifest['code_validation']['excluded_content_keys']) |
| check.close() |
| if cursor.get('exhausted_sources'): |
| raise ValueError('A required source is already exhausted; cannot preserve the fixed mixture') |
| scheduler=restore_scheduler(recipe,metadata) |
| start=recipe[KEY]['start_tokens'] |
| remaining=max(0,100_000_000_000-start) |
| capacity=readiness['capacity'] |
| if (remaining*.15>capacity['python_conservative_tokens'] |
| or remaining*.05>capacity['sql_conservative_tokens']): |
| raise ValueError('Sampled filtered supply is insufficient for the proposed 100B-total milestone') |
| result=dict(mode='resume this code phase' if resume else 'new code phase from latest full pretraining checkpoint', |
| source=str(source),output=str(OUTPUT),start_step=metadata['step'],start_tokens=metadata['tokens'], |
| transition=transition,next_lr=effective_learning_rate(metadata['tokens']+tokens_per_update(recipe),recipe,scheduler), |
| schedule='preserve existing gradual decay to 3e-5 at 100B total tokens, then hold', |
| token_shares=manifest['weights'],effective_row_limit=None, |
| data='live public Parquet streaming; bounded buffers; no corpus download or source recycling', |
| sql_order='Stack-Edu SQL first, then schema/question/query SynSQL; outer SQL allocation stays 5%', |
| optimizer='restore full FP32 AdamW state',cursor='preserve old source positions and pending input', |
| validation='same prose validation/best criterion plus separate frozen Python and SQL losses', |
| supply='sample estimates, not a full token inventory or global uniqueness guarantee',capacity=capacity) |
| return result,source,metadata,recipe,corpus |
|
|
|
|
| def correctness_probe(cluster,job): |
| """Only called after the user explicitly runs this launcher on an idle rack.""" |
| from five_mac_cluster import PYTHON,ORDER |
| probe_job=job/'correctness';probe_job.mkdir() |
| cluster.preflight(probe_job) |
| source=probe_job/'input';source.mkdir() |
| shutil.copytree(PREPARED/'runtime/sml_v1',source/'sml_v1',ignore=shutil.ignore_patterns('__pycache__','*.pyc')) |
| shutil.copyfile(ROOT/'scripts/probe_five_mac.py',source/'probe_five_mac.py') |
| cluster.stage(source) |
| cluster.launch(probe_job,source,[PYTHON,'-u',str(source/'probe_five_mac.py'),'--job-dir',str(probe_job)],timeout=360) |
| receipts=[] |
| for rank,physical in enumerate(ORDER): |
| receipt=json.loads(cluster.remote(physical,['/bin/cat',str(probe_job/f'probe_rank{rank}.json')])) |
| if receipt.get('rank')!=rank or receipt.get('status')!='passed': |
| raise RuntimeError('Missing five-rank correctness receipt') |
| receipts.append(receipt) |
| if len({r['checkpoint_digest'] for r in receipts})!=1: |
| raise RuntimeError('Five-rank replicas differ') |
| atomic_json(job/'correctness.json',dict(modules=modules(),receipts=receipts)) |
|
|
|
|
| def main(argv=None): |
| parser=argparse.ArgumentParser(description=__doc__) |
| parser.add_argument('--run',action='store_true') |
| parser.add_argument('--clear-stop',action='store_true') |
| args=parser.parse_args(argv) |
| if not args.run: |
| result,*_=plan() |
| print(json.dumps(result,indent=2)) |
| print('Plan only: no SSH, GPU work, training start or active-run changes.') |
| return result |
| |
| with lock(NETWORK/'.training-launch.lock'),lock(PARENT/'.launcher.lock'),lock(OUTPUT/'.launcher.lock'): |
| from five_mac_cluster import Cluster,PYTHON |
| from sml_v1.launch import stage |
| result,source,metadata,recipe,corpus=plan(full_verify=True) |
| if (OUTPUT/'STOP').exists() and not args.clear_stop: |
| raise ValueError('New output STOP marker exists; use --clear-stop to resume explicitly') |
| job=OUTPUT/'jobs'/(datetime.now().strftime('%Y%m%d_%H%M%S_')+uuid.uuid4().hex[:8]) |
| job.mkdir(parents=True) |
| cluster=Cluster(NETWORK) |
| cluster.preflight(job) |
| print(json.dumps(result,indent=2),flush=True) |
| atomic_json(job/'plan.json',result) |
| correctness_probe(cluster,job) |
| verify_readiness() |
| staged=stage(job,recipe,TOKENIZER,source,corpus) |
| shutil.copyfile(PREPARED/'code_validation.npz',staged/'code_validation.npz') |
| shutil.copyfile(PREPARED/'sample_prompts.json',staged/'sample_prompts.json') |
| inputs=read_json(staged/'input_manifest.json') |
| for name in ('code_validation.npz','sample_prompts.json'): |
| inputs['files'][name]=file_sha256(staged/name) |
| atomic_json(staged/'input_manifest.json',inputs) |
| if {p.name:file_sha256(p) for p in (staged/'sml_v1').glob('*.py')}!=modules(): |
| raise ValueError('Staged code differs from the prepared version') |
| cluster.stage(staged) |
| if args.clear_stop:(OUTPUT/'STOP').unlink(missing_ok=True) |
| worker=['/usr/bin/caffeinate','-i',PYTHON,'-u','-m','sml_v1.pretrain', |
| '--config',str(staged/'recipe.json'),'--tokenizer',str(staged/'tokenizer'), |
| '--corpus',str(staged/'corpus.json'),'--resume',str(staged/'checkpoint'), |
| '--save-dir',str(OUTPUT),'--job-dir',str(job),'--ring-control','native', |
| '--sample-prompts-file',str(staged/'sample_prompts.json'), |
| '--code-validation-panel',str(staged/'code_validation.npz')] |
| cluster.launch(job,staged,worker,timeout=None,stop_dir=OUTPUT) |
| result=read_json(job/'result.json') |
| if result.get('status') not in ('stopped','complete','data-exhausted'): |
| raise RuntimeError('Missing clean worker completion') |
| print(f'[verified-stop] {result}; log: {job/"train.log"}',flush=True) |
| return result |
|
|
|
|
| if __name__=='__main__':main() |
|
|
