src/common/provenance.py

"""Layered provenance hash architecture (REAL, IMPLEMENTED).

Separates hashes by semantic layer so a changed child can never hide behind an
unchanged parent: each layer hash is folded into the experiment fingerprint.
"""
import hashlib
import json
import os
from typing import Any, Dict, List, Optional


def _h(*parts) -> str:
    h = hashlib.sha256()
    for p in parts:
        h.update(str(p).encode() if not isinstance(p, bytes) else p)
        h.update(b"|")
    return h.hexdigest()


def dataset_hash(soma_path: str, connections_path: str) -> str:
    from src.connectome.loader import _file_sha256
    return _h("DATASET", _file_sha256(soma_path), _file_sha256(connections_path))


def model_hash(model_path: Optional[str]) -> str:
    from src.llm.discovery import sha256_file
    if not model_path or not os.path.exists(model_path):
        return _h("MODEL", "unavailable")
    return _h("MODEL", sha256_file(model_path))


def graph_hash(graph) -> str:
    return _h("GRAPH", graph.graph_hash)


def brain_state_hash(state) -> str:
    import numpy as np
    return _h("BRAIN",
              np.ascontiguousarray(state.membrane_potentials).tobytes(),
              np.ascontiguousarray(state.spikes).tobytes(),
              np.ascontiguousarray(state.refractory_steps).tobytes(),
              state.step_count)


def world_state_hash(world) -> str:
    return _h("WORLD", world.world_hash())


def organism_hash(organism) -> str:
    return _h("ORGANISM", organism.organism_hash())


def population_hash(pop) -> str:
    return _h("POPULATION", pop.population_hash())


def event_log_hash(event_log) -> str:
    return _h("EVENTS", event_log.compute_hash())


def experiment_fingerprint(layers: Dict[str, str]) -> str:
    """Top-level provenance fingerprint over all provided layer hashes."""
    ordered: List[str] = [f"{k}={layers[k]}" for k in sorted(layers)]
    return _h("EXPERIMENT", *ordered)


def artifact_hash(path: str) -> str:
    from src.connectome.loader import _file_sha256
    return _h("ARTIFACT", os.path.basename(path), _file_sha256(path)
              if os.path.exists(path) else "missing")


def build_layers(graph=None, state=None, world=None, population=None,
                 event_log=None, dataset=None, model_path=None) -> Dict[str, str]:
    layers: Dict[str, Any] = {}
    if dataset:
        layers["dataset"] = dataset_hash(dataset[0], dataset[1])
    if model_path is not None:
        layers["model"] = model_hash(model_path)
    if graph is not None:
        layers["graph"] = graph_hash(graph)
    if state is not None:
        layers["brain_state"] = brain_state_hash(state)
    if world is not None:
        layers["world_state"] = world_state_hash(world)
    if population is not None:
        layers["population"] = population_hash(population)
    if event_log is not None:
        layers["event_log"] = event_log_hash(event_log)
    return layers