File size: 1,808 Bytes
9118991
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
"""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}