"""Canonical JSON and content addressing for TOM World & Query Kernel 0.3.""" from __future__ import annotations import hashlib import json from typing import Any, Mapping def canonical_bytes(value: Any) -> bytes: return json.dumps( value, sort_keys=True, ensure_ascii=False, separators=(",", ":"), allow_nan=False, ).encode("utf-8") def content_hash(record: Mapping[str, Any]) -> str: body = dict(record) body.pop("content_hash", None) return "sha256:" + hashlib.sha256(canonical_bytes(body)).hexdigest() def attach_hash(record: Mapping[str, Any]) -> dict[str, Any]: out = dict(record) out.pop("content_hash", None) out["content_hash"] = content_hash(out) return out def verify_hash(record: Mapping[str, Any]) -> bool: expected = record.get("content_hash") return isinstance(expected, str) and expected == content_hash(record) def require_hash(record: Mapping[str, Any], *, label: str = "record") -> None: if not verify_hash(record): raise ValueError(f"{label} content hash mismatch")