"""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)