月度总结：把复现实验沉淀成可重放的个人研究代码库

"""Portable saved-summary capsule: integrity, two clean replays, fault injection.

Standard library only. This is a trusted-local reproducibility check, not a
sandbox for arbitrary code or a signed provenance/authenticity system.
"""
import argparse
import copy
import hashlib
import json
import os
import platform
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

ROOT = Path(__file__).resolve().parent


def digest(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()


def verify_payload(root, manifest):
    actual = {p.relative_to(root).as_posix() for p in root.rglob('*') if p.is_file()}
    if actual != set(manifest['payload']):
        raise ValueError('payload file set changed')
    for name, expected in manifest['payload'].items():
        p = root / name
        if p.is_symlink() or not p.resolve().is_relative_to(root.resolve()):
            raise ValueError('nonlocal payload path')
        if digest(p) != expected:
            raise ValueError('payload hash changed: ' + name)


def verify_environment(manifest):
    if (platform.python_version() != manifest['python'] or
            platform.python_implementation() != manifest['implementation']):
        raise ValueError('Python runtime differs from recorded baseline')


def verify_outputs(root, manifest):
    for name, expected in manifest['expected_outputs'].items():
        if not (root / name).is_file() or digest(root / name) != expected:
            raise ValueError('output differs from frozen baseline: ' + name)
    log = json.loads((root / 'verification.json').read_text())
    if log['rows'] != 60 or log['score_shape'] != [3, 5, 4]:
        raise ValueError('unexpected replay scope')
    if log['python'] != manifest['python']:
        raise ValueError('subprocess Python differs')
    return {name: digest(root / name) for name in manifest['expected_outputs']}


def fault_checks(payload, manifest, good_outputs):
    rejected = {}
    mutations = {
        'code_change': 'report.py',
        'config_change': 'input/manifest.json',
        'data_change': 'input/runs.jsonl',
        'missing_input': 'input/source_summary.json',
        'extra_file': 'unexpected.json',
    }
    for label, name in mutations.items():
        with tempfile.TemporaryDirectory(prefix='research-fault-') as td:
            target = Path(td) / 'capsule'
            shutil.copytree(payload, target)
            path = target / name
            if label == 'missing_input':
                path.unlink()
            else:
                with path.open('ab') as f:
                    f.write(b'\n# deliberate drift\n')
            try:
                verify_payload(target, manifest)
            except ValueError as exc:
                rejected[label] = str(exc)
            else:
                raise RuntimeError('undetected fault: ' + label)
    altered = copy.deepcopy(manifest)
    altered['python'] = '0.0.0'
    try:
        verify_environment(altered)
    except ValueError as exc:
        rejected['runtime_change'] = str(exc)
    else:
        raise RuntimeError('undetected runtime drift')
    with tempfile.TemporaryDirectory(prefix='research-output-fault-') as td:
        target = Path(td) / 'results'
        shutil.copytree(good_outputs, target)
        with (target / 'table.csv').open('ab') as f:
            f.write(b'corrupted result\n')
        try:
            verify_outputs(target, manifest)
        except ValueError as exc:
            rejected['output_change'] = str(exc)
        else:
            raise RuntimeError('undetected output drift')
    return rejected


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--out', type=Path, required=True,
                        help='New output directory; existing paths are refused')
    args = parser.parse_args()
    manifest = json.loads((ROOT / 'capsule_manifest.json').read_text())
    payload = ROOT / 'capsule'
    verify_environment(manifest)
    verify_payload(payload, manifest)
    out = args.out.resolve()
    out.mkdir(parents=True, exist_ok=False)
    report = {'status': 'incomplete', 'python': platform.python_version(),
              'platform': platform.platform(), 'machine': platform.machine(),
              'scope': manifest['scope'], 'runs': [],
              'manifest_sha256': digest(ROOT / 'capsule_manifest.json')}
    env = os.environ.copy()
    env['PYTHONDONTWRITEBYTECODE'] = '1'
    env['PYTHONHASHSEED'] = '81'
    env.pop('PYTHONOPTIMIZE', None)  # Copied audit deliberately uses assertions.
    try:
        for label in ('run_a', 'run_b'):
            work = out / label
            shutil.copytree(payload, work)
            verify_payload(work, manifest)
            for script in ('report.py', 'audit.py'):
                completed = subprocess.run([sys.executable, '-B', script], cwd=work,
                                           env=env, capture_output=True, text=True,
                                           timeout=60)
                (work / (script + '.stdout.txt')).write_text(completed.stdout)
                (work / (script + '.stderr.txt')).write_text(completed.stderr)
                if completed.returncode:
                    raise RuntimeError(script + ' failed with ' + str(completed.returncode))
            hashes = verify_outputs(work / 'results', manifest)
            report['runs'].append({'name': label, 'output_sha256': hashes})
        if report['runs'][0]['output_sha256'] != report['runs'][1]['output_sha256']:
            raise RuntimeError('two replays disagree')
        report['rejected_faults'] = fault_checks(payload, manifest, out / 'run_a/results')
        report['status'] = 'passed'
        report['checkpoints'] = {'payload_files': 5, 'score_shape': [3, 5, 4],
                                 'rows_per_run': 60, 'groups_per_run': 20,
                                 'stable_outputs_per_run': 5, 'clean_runs': 2}
    except Exception as exc:
        report['error'] = repr(exc)
        raise
    finally:
        (out / 'replay_audit.json').write_text(json.dumps(report, indent=2) + '\n')
        print(json.dumps(report, indent=2))


if __name__ == '__main__':
    main()