SniperGold_ML/engine/evidence.py

"""Evidence & hashing layer (Layer 11, spec 17).

Every produced artifact is listed with ``sha256_physical`` and (for canonical
layers) its logical ``content_id``. The evidence manifest hash is computed
over the manifest minus its own hash field (no self-reference / no cycles).
RUN_COMPLETE.json records the evidence manifest hash; RUN_COMPLETE.json is
never hashed by evidence.json.
"""

import os

from .canonical import (
    tick_chunk_content_id, bar_file_content_id, preservation_chunk_content_id,
)
from .storage import (
    ticks_chunk_path, bar_part_path, cert_path, chunkmap_path, journal_path,
    run_complete_path, preservation_chunk_path,
)
from .util import atomic_write_json, canonical_json_sha256, sha256_file


def _is_file(p):
    return os.path.exists(p) and os.path.isfile(p)


def _evidence_path(out_root):
    return os.path.join(out_root, "evidence", "evidence.json")


def collect_ticks_content_ids(out_root, chunkmap):
    """Logical content ids per committed ticks chunk from files on disk."""
    from .storage import read_ticks_chunk
    ids = {}
    for c in chunkmap["chunks"]:
        p = ticks_chunk_path(out_root, c["index"])
        if _is_file(p):
            rows = read_ticks_chunk(p)
            ids[c["index"]] = tick_chunk_content_id(
                [(r[0], r[1], r[2], r[4]) for r in rows])
    return ids


def collect_preservation_content_ids(out_root, chunkmap):
    """Logical content ids per committed source-preservation sidecar
    (spec 8.6). Re-reads the sidecar lines ``src_line|last_u`` from disk."""
    ids = {}
    for c in chunkmap["chunks"]:
        p = preservation_chunk_path(out_root, c["index"])
        if not _is_file(p):
            continue
        with open(p, "r", encoding="utf-8") as fh:
            pairs = []
            for line in fh:
                line = line.rstrip("\r\n")
                if not line:
                    continue
                src_line, last_u = line.split("|", 1)
                pairs.append((int(src_line), int(last_u)))
        ids[c["index"]] = preservation_chunk_content_id(pairs)
    return ids


def collect_bar_content_ids(out_root, chunkmap, cfg):
    from .storage import read_bar_part
    ids = {}
    for wl in chunkmap["workloads"]:
        for tf in cfg["timeframes"]:
            p = bar_part_path(out_root, tf, wl["index"])
            if _is_file(p):
                rows = read_bar_part(p)
                ids[(tf, wl["index"])] = bar_file_content_id(rows)
    return ids


def collect_file_records(out_root, chunkmap, cfg=None):
    """Enumerate engine-output files with physical + logical hashes.

    Non-canonical artifacts (cert, chunkmap, malformed sidecars, checkpoint,
    journal, config snapshot) record content_id = null -- no canonical
    serialization applies (documented). Ticks/bar canonical layers record
    the logical content id per spec 8.5 / 12.4.
    """
    records = {}

    def add(relpath):
        full = os.path.join(out_root, relpath)
        if not _is_file(full):
            return
        records[relpath] = {
            "bytes": os.path.getsize(full),
            "sha256_physical": sha256_file(full),
            "content_id": _canonical_or_physical(out_root, relpath, chunkmap, cfg),
        }

    cid_ticks = collect_ticks_content_ids(out_root, chunkmap)
    for c in chunkmap["chunks"]:
        add("ticks/chunk_%06d.parquet" % c["index"])
    if cid_ticks:
        for rel, rec in records.items():
            if rel.startswith("ticks/"):
                idx = int(rel.split("_")[-1].split(".")[0])
                rec2 = dict(rec)
                rec2["content_id"] = cid_ticks.get(idx)
                records[rel] = rec2
    bar_ids = collect_bar_content_ids(out_root, chunkmap, cfg) if cfg else {}
    for wl in chunkmap["workloads"]:
        if cfg is not None:
            for tf in cfg["timeframes"]:
                rel = "bars/%s/wl_%03d.parquet" % (tf, wl["index"])
                add(rel)
                if rel in records and (tf, wl["index"]) in bar_ids:
                    records[rel]["content_id"] = bar_ids[(tf, wl["index"])]
    for c in chunkmap["chunks"]:
        add("malformed/malformed_%06d.jsonl" % c["index"])
    pres_ids = collect_preservation_content_ids(out_root, chunkmap)
    for c in chunkmap["chunks"]:
        rel = "source_preservation/chunk_%06d.jsonl" % c["index"]
        add(rel)
        if rel in records and c["index"] in pres_ids:
            records[rel]["content_id"] = pres_ids[c["index"]]
    add("certification/source_certificate.json")
    add("chunkmap/chunkmap-%s.json" % chunkmap["source_id"])
    add("state/checkpoint.json")
    add("state/commits.jsonl")
    return records


def _canonical_or_physical(out_root, relpath, chunkmap, cfg):
    if relpath.startswith(("ticks/", "bars/")):
        # caller supplies the logical id via the post-pass below; placeholder
        return None
    if relpath == "state/commits.jsonl":
        return None
    return None


def build_evidence(out_root, chunkmap, cfg=None):
    records = collect_file_records(out_root, chunkmap, cfg)
    payload = {
        "schema_version": "EVIDENCE_V1",
        "files": records,
    }
    payload["evidence_manifest_hash"] = canonical_json_sha256(
        {k: v for k, v in payload.items() if k != "evidence_manifest_hash"})
    atomic_write_json(_evidence_path(out_root), payload)
    return payload


def load_evidence(out_root):
    import json
    with open(_evidence_path(out_root), "r", encoding="utf-8") as fh:
        return json.load(fh)


def verify_evidence(out_root):
    """Recompute every entry from disk; physical hashes must match and the
    evidence manifest hash must be stable. Returns (ok, diffs)."""
    payload = load_evidence(out_root)
    body = {k: v for k, v in payload.items() if k != "evidence_manifest_hash"}
    expected = payload.get("evidence_manifest_hash")
    if canonical_json_sha256(body) != expected:
        return False, ["evidence_manifest_hash mismatch"]
    diffs = []
    for relpath, rec in payload.get("files", {}).items():
        full = os.path.join(out_root, relpath)
        if not _is_file(full):
            diffs.append("missing:" + relpath)
            continue
        if rec.get("bytes") != os.path.getsize(full):
            diffs.append("size:" + relpath)
        if rec.get("sha256_physical") != sha256_file(full):
            diffs.append("sha256:" + relpath)
    return (not diffs), diffs


def evidence_manifest_hash(evidence):
    return evidence.get("evidence_manifest_hash")


def run_complete_path_of(out_root):
    return run_complete_path(out_root)