"""Export verified completed cohorts; never turn partial journals into full rates.""" import argparse from collections import Counter import gzip import hashlib import json import math from pathlib import Path import statistics import sys def export(root, historical, harness, output): sys.path.insert(0, str(harness)) from rl_review_summary import split_statistics, distribution hashes = {} def read(path): raw = path.read_bytes() hashes[str(path)] = hashlib.sha256(raw).hexdigest() return json.loads(raw) cohort = root/'results/review-resumed-01' evaluations = root/'results/review-evaluation-04' strict_report = read(evaluations/'summary.json') assert strict_report['schema'] == 'simcompare.review-cohort-summary.v1' protocol = read(cohort/'protocol.json') assert protocol['seeds'] == list(range(5)) and protocol['test_starts'] == 58 assert protocol['force_penalties'] == [0., 10.] selections = {} for weight in (0, 10): for seed in range(5): run = cohort/'runs'/f'insert-force{weight}-seed{seed}' status = read(run/'status.json') assert status['status'] == 'complete' and status['environment_steps'] == 262144 pointer = read(run/'best-checkpoint.json') checkpoint = run/pointer['checkpoint'] manifest = read(checkpoint.parent/'checkpoint-manifest.json') assert manifest['environment_steps'] == 262144 assert set(manifest['files']) == {'policy.zip', 'normalization.json', 'interface.json'} for name, digest in manifest['files'].items(): assert hashlib.sha256((checkpoint.parent/name).read_bytes()).hexdigest() == digest assert len(read(checkpoint.parent/'validation-outcomes.json')) == 24 selections[f'{weight}:{seed}'] = dict(seed=seed, force_weight=weight, selected_steps=manifest['environment_steps'], files=manifest['files'], validation=pointer['validation']) engines, episodes, pending = {}, [], [] for engine in ('mujoco_cpu', 'newton_mjwarp', 'physx_gpu', 'genesis'): engines[engine] = {} for weight in (0, 10): directory = evaluations/f'force{weight}-{engine}' if not (directory/'status.json').exists(): assert (engine, weight) == ('genesis', 10), 'Unexpected incomplete cohort' journal = directory/'episodes.jsonl' count = len(journal.read_text().splitlines()) if journal.exists() else 0 pending.append(dict(engine=engine, weight=weight, preserved_at_restart=count, expected_episodes=290, reason='Interrupted by service restart; continuation running. No final rate exported.')) continue status, metadata = read(directory/'status.json'), read(directory/'evaluation.json') expected_count = 348 if weight == 0 else 290 assert status['status'] == 'complete' and not status['errors'] assert status['episodes'] == expected_count assert metadata['backend'] == engine and metadata['scene_indices'] == list(range(58)) assert metadata['reference'] == (weight == 0) and metadata['common_evaluation_force_penalty'] == 0. rows = read(directory/'outcomes.json') assert len(rows) == expected_count assert len({(r['scene_index'], r['seed'], r['policy']) for r in rows}) == expected_count labels = {} for policy in metadata['policies']: matches = [s for s in range(5) if f'insert-force{weight}-seed{s}' in Path(policy['label']).parts] assert len(matches) == 1 and matches[0] not in labels.values() seed = matches[0] assert policy['sha256'] == selections[f'{weight}:{seed}']['files']['policy.zip'] labels[policy['label']] = seed assert set(labels.values()) == set(range(5)) assert {r['policy'] for r in rows} == set(labels) | ({'alignment_reference'} if weight == 0 else set()) seed_results = [] for label, seed in sorted(labels.items(), key=lambda x: x[1]): selected = [r for r in rows if r['policy'] == label] assert all(r['backend'] == engine and r['training_force_penalty'] == weight and r['policy_sha256'] == selections[f'{weight}:{seed}']['files']['policy.zip'] for r in selected) seed_results.append(dict(seed=seed, subsets=split_statistics(selected, protocol['conditions']['test']))) result = dict(status='complete', episodes=expected_count, seeds=seed_results, success_rate_across_seeds={subset: distribution([r['subsets'][subset]['success_rate'] for r in seed_results]) for subset in ('new_50', 'prior_8', 'all_58')}) assert {r['seed']: r['subsets'] for r in seed_results} == { r['training_seed']: r['subsets'] for r in strict_report['engines'][engine][str(weight)]['seeds']} if weight == 0: result['reference'] = split_statistics([r for r in rows if r['policy'] == 'alignment_reference'], protocol['conditions']['test']) if 'elapsed_s' in status: result['evaluation_wall_s'] = status['elapsed_s'] engines[engine][str(weight)] = result for row in rows: episodes.append(dict(engine=engine, force_weight=weight, training_seed=labels.get(row['policy']), reference=row['policy'] == 'alignment_reference', scene_index=row['scene_index'], partition='prior_8' if row['scene_index'] < 8 else 'new_50', evaluation_seed=row['seed'], policy_sha256=row['policy_sha256'], **{k:row.get(k) for k in ('success', 'outcome', 'failure', 'physics_steps', 'peak_force_N', 'peak_moment_Nm', 'metrics', 'replay')})) assert not pending and len(episodes) == 2552 assert sum(len(v) for v in engines.values()) == 8 audits = {} for path in sorted((historical/'results').glob('review-genesis-part1-*/**/audit.json')): value = read(path) assert value['forbidden_contact_samples'] == 0 audits[path.parent.name] = dict(path=path, value=value) assert len(audits) == 14 pairs = [] for name in sorted(k[:-6] for k in audits if k.endswith('-local')): local, global_ = audits[name+'-local'], audits[name+'-global'] traces = [] for item in (local, global_): path = item['path'].parent/'trace.json.gz' raw = path.read_bytes(); hashes[str(path)] = hashlib.sha256(raw).hexdigest() traces.append(json.loads(gzip.decompress(raw))) a, b = traces assert len(a['frames']) == len(b['frames']) maximum = 0. for fa, fb in zip(a['frames'], b['frames']): assert fa['time_s'] == fb['time_s'] bodies = {x['id']:x for x in fb['bodies']} for x in fa['bodies']: y = bodies[x['id']] maximum = max(maximum, math.dist(x['position_m'], y['position_m'])) pairs.append(dict(case=name, max_body_position_difference_m=maximum, local_metrics=local['value']['metrics'], global_metrics=global_['value']['metrics'])) audit_export = dict(runs=14, forbidden_contact_samples=0, paired_conditions=pairs, limitation='Original-controller checks only; 80 mm remains sensitive in trajectory despite the same failed task outcome. Not a full replay parity or repeatability study.') result = dict(schema='simcompare.blog.five-seed-study.v1', date='2026-10-09', status='partial' if pending else 'complete', completed_cohorts=sum(len(v) for v in engines.values()), completed_evaluation_episodes=len(episodes), pending=pending, scope='All ten policies trained in MuJoCo CPU. Five seeds per reward, 50 new tests and eight prior cases; completed cohorts only. Earlier pilots and diagnostic/visualization repeats excluded.', protocol=protocol, selections=selections, engines=engines, contrasts={engine:{part:distribution([ conditions['10']['seeds'][seed]['subsets'][part]['success_rate'] - conditions['0']['seeds'][seed]['subsets'][part]['success_rate'] for seed in range(5)]) for part in ('new_50','prior_8','all_58')} for engine,conditions in engines.items() if '10' in conditions}, recovery=read(cohort/'recovery.json'), genesis_collision_audit=audit_export, concurrency_probe=read(root/'results/review-parallel-probe-01/comparison.json'), source_files_sha256=hashes, analysis_source_sha256=hashlib.sha256((harness/'rl_review_summary.py').read_bytes()).hexdigest()) output.mkdir(parents=True, exist_ok=True) for name,value in [('summary.json',result),('episodes.json',episodes)]: (output/name).write_text(json.dumps(value,indent=2,allow_nan=False)+'\n') print(json.dumps(dict(completed_cohorts=result['completed_cohorts'], completed_episodes=len(episodes), pending=pending, collision_pairs=pairs),indent=2)) if __name__ == '__main__': parser=argparse.ArgumentParser() for name in ('root','historical','harness','output'): parser.add_argument('--'+name,type=Path,required=True) args=parser.parse_args() export(args.root,args.historical,args.harness,args.output)