Buckets:
| """Audited model-first relocation of historical experiment run directories.""" | |
| from __future__ import annotations | |
| import argparse | |
| import collections | |
| import json | |
| import os | |
| from pathlib import Path | |
| import re | |
| import yaml | |
| STAMP = re.compile(r"^\d{8}-\d{6}_") | |
| CATEGORIES = {"runs", "smoke", "sweeps", "archive"} | |
| MODELS = { | |
| "uniar-sft", "uniar-rl", "bagel-7b-mot", "thinkmorph-7b", "latentum-base", | |
| "anole-zebra-cot", "bagel-zebra-cot", "qwen3-vl-8b-instruct", | |
| "qwen3.5-9b", "qwen3.8-27b", | |
| } | |
| METADATA = ( | |
| "run.json", "config.json", "config.yaml", "arguments.json", "args.json", | |
| "effective_args.json", "model_manifest.json", "shard_00_config.json", | |
| "_submission/argv.json", "summary.json", "model_config.json", "request.json", "results.json", "cache/config.json", | |
| ) | |
| MODEL_KEYS = { | |
| "model_path", "base_model_path", "model_id", "model_name", "model_name_or_path", | |
| "pretrained_model_name_or_path", "model_label", "model_type", "base_model", "model_dir", | |
| } | |
| def run_roots(root: Path): | |
| for current, dirs, _ in os.walk(root, followlinks=False): | |
| path = Path(current) | |
| if STAMP.match(path.name): | |
| yield path | |
| dirs.clear() | |
| continue | |
| dirs[:] = [name for name in dirs if not name.startswith(('.', '_')) | |
| and not (path == root and name in MODELS | {"ops", "shared"})] | |
| def read_metadata(path: Path): | |
| if not path.is_file() or path.stat().st_size > 2_000_000: | |
| return None | |
| try: | |
| text = path.read_text(encoding="utf-8") | |
| return yaml.safe_load(text) if path.suffix == '.yaml' else json.loads(text) | |
| except (ValueError, OSError, yaml.YAMLError): | |
| return None | |
| def model_slug(value: str) -> str | None: | |
| name = value.lower().replace('_', '-').replace('qwen3-8', 'qwen3.8').replace('qwen3-5', 'qwen3.5') | |
| for marker, slug in ( | |
| ('uniar-rl', 'uniar-rl'), ('uniar', 'uniar-sft'), | |
| ('bagel-zebra', 'bagel-zebra-cot'), ('anole', 'anole-zebra-cot'), | |
| ('qwen3.8', 'qwen3.8-27b'), ('qwen3.5', 'qwen3.5-9b'), | |
| ('qwen3-vl', 'qwen3-vl-8b-instruct'), ('latentum', 'latentum-base'), | |
| ('thinkmorph-7b', 'thinkmorph-7b'), ('thinkmorph7b', 'thinkmorph-7b'), | |
| ('bagel-7b', 'bagel-7b-mot'), | |
| ): | |
| if marker in name: | |
| return slug | |
| return None | |
| def primary_models(payload, *, parent="") -> set[str]: | |
| result = set() | |
| if not isinstance(payload, dict): | |
| return result | |
| for key, value in payload.items(): | |
| if isinstance(value, str) and (key in MODEL_KEYS or | |
| (parent == 'model' and key in {'path', 'variant', 'name', 'id'}) or key == 'model'): | |
| slug = model_slug(value) | |
| if slug: | |
| result.add(slug) | |
| elif isinstance(value, dict) and key in {'config', 'framework', 'model', 'args', 'arguments'}: | |
| result.update(primary_models(value, parent=key)) | |
| return result | |
| def identity(path: Path, root: Path) -> tuple[str | None, list[str]]: | |
| parts = path.relative_to(root).parts | |
| top = parts[0] | |
| if top in {'bagel', 'latentum'}: | |
| return parts[1], ['existing backend/model namespace'] | |
| if top in {'cluster', 'uniar_environment'}: | |
| return 'ops', ['operational output namespace'] | |
| if top in {'site-rollouts', 'figures'}: | |
| return 'shared', ['presentation/derived artifact namespace'] | |
| # A joint comparison run is kept as one unit, including shared metrics. | |
| if all((path / name).is_dir() for name in ('bagel', 'anole')): | |
| return 'shared', ['joint BAGEL/Anole comparison run'] | |
| found: dict[str, list[str]] = collections.defaultdict(list) | |
| argv = [] | |
| for relative in METADATA: | |
| payload = read_metadata(path / relative) | |
| for model in primary_models(payload): | |
| found[model].append(relative) | |
| if relative == '_submission/argv.json' and isinstance(payload, dict): | |
| argv = payload.get('original', []) | |
| for index, value in enumerate(argv): | |
| if not isinstance(value, str): | |
| continue | |
| candidate = None | |
| if value.startswith(('MODEL_NAME=', 'MODEL_PATH=', 'BASE_MODEL=')): | |
| candidate = value.split('=', 1)[1] | |
| if index and argv[index - 1] in {'--model-path', '--model-name', '--model-label', '--model'}: | |
| candidate = value | |
| if candidate and (slug := model_slug(candidate)): | |
| found[slug].append('_submission/argv.json:model argument') | |
| if len(found) == 1: | |
| owner = next(iter(found)) | |
| return owner, sorted(set(found[owner])) | |
| if len(found) > 1: | |
| return None, ['conflicting primary model metadata: ' + ','.join(sorted(found))] | |
| launch = ' '.join(str(value) for value in argv) | |
| if 'cluster_jobs.smoke' in launch: | |
| return 'ops', ['submission infrastructure smoke entrypoint'] | |
| if 'ThinkMorph-OPD/' in launch or 'thinkmorph_native' in launch: | |
| return 'thinkmorph-7b', ['native ThinkMorph launch entrypoint'] | |
| if 'OURS/config/' in launch or 'OURS/tests/test_uniar' in launch or any('zebra_sft.' in value for value in argv if isinstance(value, str)): | |
| return 'uniar-sft', ['UniAR-only launch module/config'] | |
| slug = model_slug(path.name) | |
| if slug: | |
| return slug, ['explicit model in recorded run name (metadata absent)'] | |
| if top.startswith('uniar_'): | |
| return 'uniar-sft', ['UniAR-only diagnostic namespace'] | |
| # Older runs sometimes kept only shell launchers, reports, or log headers. | |
| text_models: dict[str, list[str]] = collections.defaultdict(list) | |
| candidates = sorted(path.iterdir()) + [path / '_submission/job.sbatch'] | |
| for candidate in candidates: | |
| if not candidate.is_file() or candidate.suffix not in {'.json', '.jsonl', '.sh', '.py', '.md', '.out', '.log', '.sbatch', '.tsv'}: | |
| continue | |
| if candidate.name == 'environment.json': | |
| continue | |
| with candidate.open('r', errors='replace') as handle: | |
| text = handle.read(131072) | |
| for match in re.finditer(r'(?:models/|MODEL_NAME=|--model(?:-path|-name)?[ =]+)([^\s\"\x27;,]+)', text): | |
| if slug := model_slug(match.group(1)): | |
| text_models[slug].append(str(candidate.relative_to(path))) | |
| if re.search(r'(?:from|import) (?:uniar|zebra_sft)\b|(?:-m |src/)(?:zebra_sft|sft\b)', text): | |
| text_models['uniar-sft'].append(str(candidate.relative_to(path)) + ': UniAR module') | |
| if len(text_models) == 1: | |
| owner = next(iter(text_models)) | |
| return owner, sorted(set(text_models[owner])) | |
| if len(text_models) > 1: | |
| return 'shared', ['multi-model launch/report: ' + ','.join(sorted(text_models))] | |
| return None, ['model not resolved from primary metadata or launch arguments'] | |
| def destination(path: Path, root: Path, owner: str) -> Path: | |
| parts = path.relative_to(root).parts | |
| top = parts[0] | |
| if top in {'bagel', 'latentum'}: | |
| return root.joinpath(owner, *parts[2:]) | |
| topic = top.replace('_', '-') | |
| version = next((part for part in parts[1:-1] if re.fullmatch(r'(?:v|RLv)\d+', part)), 'unversioned') | |
| category_index = next((i for i, part in enumerate(parts[:-1]) if part in CATEGORIES), None) | |
| if category_index is None: | |
| category = 'smoke' if 'smoke' in path.name else ('sweeps' if 'sweep' in path.name else 'runs') | |
| task = 'eval' if top == 'cv_bench' else 'diagnostics' | |
| else: | |
| category = parts[category_index] | |
| between = parts[category_index + 1:-1] | |
| if len(between) > 1: | |
| raise ValueError(f'unexpected nested task namespace: {path}') | |
| task = between[0] if between else ('eval' if 'qwen' in path.name else 'diagnostics') | |
| if top.startswith('uniar_'): | |
| topic = 'environment' if top == 'uniar_environment' else 'runtime' | |
| task = top.removeprefix('uniar_').replace('_', '-') | |
| if top == 'figures': | |
| task = 'render' | |
| return root / owner / topic / version / category / task / path.name | |
| def plan(root: Path, overrides: dict | None = None) -> dict: | |
| rows = [] | |
| destinations = set() | |
| paths = sorted(run_roots(root)) | |
| identities = {path: identity(path, root) for path in paths} | |
| names = collections.defaultdict(set) | |
| for path, (owner, _) in identities.items(): | |
| if owner and owner not in {'ops', 'shared'}: | |
| names[path.name].add(owner) | |
| reference_pattern = re.compile('|'.join(re.escape(name) for name in sorted(names, key=len, reverse=True))) | |
| for path in paths: | |
| owner, evidence = identities[path] | |
| if owner is None: | |
| references = set() | |
| for file in sorted(path.iterdir()) + [path / '_submission/argv.json']: | |
| if file.is_file() and file.suffix in {'.json', '.jsonl', '.tsv', '.md', '.sh'}: | |
| with file.open(errors='replace') as handle: | |
| text = handle.read(262144) | |
| for name in reference_pattern.findall(text): | |
| references.update(names[name]) | |
| if references: | |
| owner = next(iter(references)) if len(references) == 1 else 'shared' | |
| evidence = ['references classified runs: ' + ','.join(sorted(references))] | |
| override = (overrides or {}).get(path.name) | |
| if override: | |
| owner, evidence = override['owner'], [override['evidence']] | |
| if owner not in MODELS | {'ops', 'shared'}: | |
| raise ValueError(f'invalid reviewed owner: {owner}') | |
| target = destination(path, root, owner) if owner else None | |
| if target and override and override.get('unclassified'): | |
| target = root / 'shared' / 'unclassified' / Path(*target.relative_to(root).parts[2:]) | |
| if target is not None: | |
| if target in destinations or target.exists(): | |
| raise FileExistsError(f'existing/duplicate destination: {target}') | |
| destinations.add(target) | |
| rows.append(dict(source=str(path.relative_to(root)), | |
| destination=str(target.relative_to(root)) if target else None, | |
| owner=owner, evidence=evidence)) | |
| return dict(schema='spatial_mmcot_output_layout_migration_v1', output_root=str(root.resolve()), records=rows) | |
| def main(): | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument('--output-root', type=Path, default=Path('experiments/output')) | |
| parser.add_argument('--plan', type=Path, required=True) | |
| parser.add_argument('--overrides', type=Path) | |
| args = parser.parse_args() | |
| overrides = read_metadata(args.overrides) if args.overrides else None | |
| result = plan(args.output_root, overrides) | |
| args.plan.parent.mkdir(parents=True, exist_ok=True) | |
| with args.plan.open('x', encoding='utf-8') as out: | |
| json.dump(result, out, indent=2) | |
| out.write('\n') | |
| print(json.dumps(dict(records=len(result['records']), | |
| owners=dict(collections.Counter(r['owner'] for r in result['records']))))) | |
| for record in result['records']: | |
| if record['owner'] is None: | |
| print('UNRESOLVED', record['source'], record['evidence']) | |
| if __name__ == '__main__': | |
| main() | |
Xet Storage Details
- Size:
- 11.2 kB
- Xet hash:
- d70fe371da2e3e18f47299440f4912822793b7f955934b89dbc01fcc721c2bf6
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.