Files
MeanField/reports/archive/organize_artifacts.py
Emily Boudreaux e7e53ce682 Consolidate numerical research and preserve a reproducible handoff
Document the baseline, hypotheses, measured results, failed approaches,
current policy defaults, and remaining rotating n=3 and virial limitations.
Include the authoritative LaTeX report, Pandoc-generated Markdown, compiled
PDF, source provenance, branch decisions, and consolidation validation.

Organize raw results and miscellaneous artifacts under working/ without
deleting them; version the directory guide, relocation hashes, archive
verifier, and report build/conversion checks. Keep build caches at their
original paths and clearly distinguish retained binaries from the validated
research build. Preserve experimental policies without promoting failed
trials to defaults.
2026-09-20 09:10:27 -04:00

68 lines
3.0 KiB
Python

"""Execute the reviewed move plan without deleting or rewriting evidence."""
from pathlib import Path
import hashlib
import json
import os
import subprocess
import time
HERE = Path(__file__).resolve().parent
REPO = Path('/Users/tboudreaux/Programming/4DSTAR/MeanField')
def inventory(path):
entries = []
paths = [path] if not path.is_dir() else sorted(path.rglob('*'))
for item in paths:
relative = '.' if item == path else str(item.relative_to(path))
if item.is_symlink():
entries.append({'path': relative, 'kind': 'symlink', 'target': os.readlink(item)})
elif item.is_file():
digest = hashlib.sha256()
with item.open('rb') as stream:
for block in iter(lambda: stream.read(1024 * 1024), b''):
digest.update(block)
entries.append({'path': relative, 'kind': 'file', 'bytes': item.stat().st_size,
'sha256': digest.hexdigest()})
elif item.is_dir():
entries.append({'path': relative, 'kind': 'directory'})
else:
raise RuntimeError(f'Unsupported evidence type: {item}')
return entries
plan = json.loads((HERE / 'move_plan.json').read_text())
tracked = [REPO / x for x in subprocess.check_output(
['git', 'ls-files', '-z'], cwd=REPO).decode().split('\0') if x]
for move in plan:
source, destination = Path(move['source']), Path(move['destination'])
if not source.exists() or source.is_symlink():
raise RuntimeError(f'Missing/already-moved source: {source}')
if destination.exists() or destination.is_symlink():
raise RuntimeError(f'Destination already exists: {destination}')
if any(x == source or source in x.parents for x in tracked):
raise RuntimeError(f'Plan would move tracked source: {source}')
manifest = {'started_at_unix': time.time(), 'moves': []}
record = HERE / 'move_manifest.json'
for move in plan:
source, destination = Path(move['source']), Path(move['destination'])
before = inventory(source)
entry = {**move, 'entries': before, 'verified': False}
manifest['moves'].append(entry)
record.write_text(json.dumps(manifest, indent=2) + '\n')
destination.parent.mkdir(parents=True, exist_ok=True)
# Atomic rename preserves contents, permissions and directory structure.
source.rename(destination)
if move['compatibility_symlink']:
source.symlink_to(destination, target_is_directory=destination.is_dir())
if inventory(destination) != before:
raise RuntimeError(f'Post-move evidence mismatch: {destination}')
entry['verified'] = True
record.write_text(json.dumps(manifest, indent=2) + '\n')
print(f'Preserved and verified: {source.name} ({len(before)} entries)', flush=True)
manifest['completed_at_unix'] = time.time()
manifest['all_verified'] = all(x['verified'] for x in manifest['moves'])
record.write_text(json.dumps(manifest, indent=2) + '\n')
print('All moves verified; no artifacts deleted.', flush=True)