loopq_quantization/scripts/loopq/evaluation_provenance.py

"""Record evaluation source bytes and detect edits during an external run."""
from __future__ import annotations

import hashlib
from pathlib import Path

SOURCE_AREAS = ('loopQ', 'src', 'scripts', 'plugins/ouro_quant_vllm',
                'plugins/huginn_vllm_validated')
SOURCE_SUFFIXES = {'.py', '.yaml', '.yml', '.json', '.toml', '.cu', '.cuh', '.cpp', '.h'}


def source_paths(root):
    for area in SOURCE_AREAS:
        for path in sorted((root / area).rglob('*')):
            if path.is_file() and path.suffix in SOURCE_SUFFIXES and '__pycache__' not in path.parts:
                yield path


def capture_evaluation_sources(root: Path, output: Path) -> dict:
    destination = output / 'source_snapshot'
    destination.mkdir(parents=True, exist_ok=False)
    hashes = {}
    for path in source_paths(root):
        relative = path.relative_to(root)
        data = path.read_bytes()
        hashes[str(relative)] = hashlib.sha256(data).hexdigest()
        target = destination / relative
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_bytes(data)
    return {'snapshot': str(destination.resolve()), 'sha256_at_launch': hashes,
            'scope': 'repository source at launch; installed dependencies are identified separately by environment metadata'}


def finish_evaluation_provenance(root: Path, provenance: dict) -> dict:
    current = {str(path.relative_to(root)): hashlib.sha256(path.read_bytes()).hexdigest()
               for path in source_paths(root)}
    previous = provenance['sha256_at_launch']
    changed = sorted(key for key in set(previous) | set(current)
                     if previous.get(key) != current.get(key))
    return provenance | {'changed_during_run': changed,
                         'source_tree_unchanged_at_exit': not changed}