hashing.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/source/hashing.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/source/test_hashing.py"
"""
The three AuthoredSource hash projections (+ envelope nullability).
 
Purpose
-------
Implement Registry Source & Storage Spec V0.22 §3.2: (1) the canonical
source-tree hash, (2) the source semantic projection, (3) the canonical
logical-state hash over the RuntimeStore durable roadmap rows (surrogate →
natural-key resolution + canonical ``query_params_json``), plus the §10 raw
source-tree hash and the failure-envelope hash nullability rule.
 
Responsibilities
----------------
- ``raw_source_tree_hash`` — sha256 over raw observed bytes (incl. files that
  later fail validation; computable before parse; null only on scan failure).
- ``source_hash`` — sha256 over canonical record bytes (projection 1).
- ``semantic_hash`` — sha256 over the source semantic projection (projection 2).
- ``logical_state_hash`` — sha256 over the §3.2 store projection (projection 3).
- ``hash_preimage`` — the printed preimage (with ``hash_preimage_schema_version``).
- ``stage_hashes`` — the §10 nullability of the three hashes per failure stage.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - hash lifecycle
    L3 - projection details
    L4 - per-row projection tracing
 
Contracts
---------
- All hashes are sha256 hex over ``canonical_json`` of a preimage that begins
  with ``hash_preimage_schema_version`` (starts at integer 1).
- The logical projection includes/excludes exactly the §3.2 columns; surrogate
  keys are resolved to natural keys; ``query_params_json`` is re-emitted canonical.
- ``logical_state_hash`` reads an already-built store; it derives nothing
  (version status/lineage derive-verify is the P1.4 step-7 publish-gate).
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.canonical (canonical_json)
- oc_chat.registry.source.scanner (ScannedRecord and observed_files - the records and files the hashes cover)
- oc_chat.registry.source.canonical (canonical_record_bytes - the canonical bytes behind the source hash)
- oc_chat.registry.source.errors (SRC_BAD_QUERY_PARAMS and SourceValidationError - an invalid query_params_json)
- option_c_registry_contracts (RegistryModel - the loaded record models behind the semantic hash)
"""
from __future__ import annotations
 
import hashlib
import json
import sqlite3
from pathlib import Path
from typing import Any, Optional
 
from oc_diagnostics import _dbg
 
from option_c_registry_contracts import RegistryModel
 
from oc_chat.registry.canonical import canonical_json
from oc_chat.registry.source.canonical import canonical_record_bytes
from oc_chat.registry.source.errors import (
    SRC_BAD_QUERY_PARAMS,
    SourceValidationError,
)
from oc_chat.registry.source.scanner import ScannedRecord, observed_files
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
#: Bumps only when a projection definition changes (§3.2; Opus 4.7 V9 L-V9-1).
HASH_PREIMAGE_SCHEMA_VERSION = 1
 
# Failure stages for the §10 envelope hash nullability rule (AC §8.48).
STAGE_SCAN = "scan"
STAGE_PARSE = "parse"
STAGE_CROSS_RECORD = "cross_record"
STAGE_BUILD = "build"
STAGE_CLEAN = "clean"
 
 
def _sha256_hex(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()
 
 
def _hash_preimage(preimage: dict[str, Any]) -> str:
    return _sha256_hex(canonical_json(preimage).encode("utf-8"))
 
 
# --- Projection 1 (raw + canonical source-tree) -----------------------------
def raw_source_tree_preimage(root: Path) -> dict[str, Any]:
    """``[(relative_posix_path, sha256(raw_file_bytes))]`` over observed files."""
    entries = [
        [path.relative_to(root).as_posix(), _sha256_hex(path.read_bytes())]
        for path in observed_files(Path(root))
    ]
    entries.sort()
    return {
        "hash_preimage_schema_version": HASH_PREIMAGE_SCHEMA_VERSION,
        "files": entries,
    }
 
 
def raw_source_tree_hash(root: Path) -> str:
    return _hash_preimage(raw_source_tree_preimage(root))
 
 
def source_hash_preimage(records: list[ScannedRecord]) -> dict[str, Any]:
    """``[(relative_posix_path, sha256(canonical_record_bytes))]`` (projection 1).
 
    [OP-5] promoted parity: promoted items are now included (the earlier
    model-gap skip is removed), so ``source_hash`` is a complete fingerprint of
    the authored tree — an edit to a promoted file moves ``source_hash``.
    """
    entries: list[list[str]] = []
    for scanned in records:
        entries.append(
            [scanned.rel_path, _sha256_hex(canonical_record_bytes(scanned))]
        )
    entries.sort()
    return {
        "hash_preimage_schema_version": HASH_PREIMAGE_SCHEMA_VERSION,
        "records": entries,
    }
 
 
def source_hash(records: list[ScannedRecord]) -> str:
    return _hash_preimage(source_hash_preimage(records))
 
 
# --- Projection 2 (source semantic) -----------------------------------------
def semantic_preimage(loaded: list[tuple[ScannedRecord, RegistryModel]]) -> dict[str, Any]:
    """Authored records normalized to canonical JSON using source field names."""
    records = [
        {
            "kind": scanned.kind,
            "rel_path": scanned.rel_path,
            "fields": model.model_dump(mode="json"),
        }
        for scanned, model in loaded
    ]
    records.sort(key=lambda entry: entry["rel_path"])
    return {
        "hash_preimage_schema_version": HASH_PREIMAGE_SCHEMA_VERSION,
        "records": records,
    }
 
 
def semantic_hash(loaded: list[tuple[ScannedRecord, RegistryModel]]) -> str:
    return _hash_preimage(semantic_preimage(loaded))
 
 
# --- Projection 3 (canonical logical-state) ---------------------------------
def _table_exists(conn: sqlite3.Connection, name: str) -> bool:
    row = conn.execute(
        # OCT-LINT: disable=no_hardcoded_secrets reason="Column/table names are schema constants, not secrets"
        "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (name,)
    ).fetchone()
    return row is not None
 
 
def _rows(conn: sqlite3.Connection, table: str, columns: str) -> list[sqlite3.Row]:
    if not _table_exists(conn, table):
        return []
    return conn.execute(f"SELECT {columns} FROM {table}").fetchall()
 
 
def _canonical_query_params(value: Optional[str], section_path: str) -> Optional[str]:
    if value is None or value == "":
        return None
    try:
        parsed = json.loads(value)
    except (ValueError, TypeError) as error:
        raise SourceValidationError(
            f"render section {section_path!r}: invalid query_params_json: {error}",
            code=SRC_BAD_QUERY_PARAMS,
            records=[section_path],
        ) from error
    return canonical_json(parsed)
 
 
def logical_state_projection(conn: sqlite3.Connection) -> dict[str, Any]:
    """Build the §3.2 projection-3 preimage over the durable roadmap rows."""
    conn.row_factory = sqlite3.Row
    phase_code_by_id = {
        row["phase_id"]: row["phase_code"]
        for row in _rows(conn, "roadmap_phases", "phase_id, phase_code")
    }
    track_code_by_id = {
        row["track_id"]: row["track_code"]
        for row in _rows(conn, "roadmap_tracks", "track_id, track_code")
    }
 
    eras = [
        dict(row)
        for row in _rows(
            conn,
            "roadmap_eras",
            "era_id, era_label, major_bump_reason, archived_under, "
            "sort_order, started_at, ended_at",
        )
    ]
    eras.sort(key=lambda row: row["era_id"])
 
    versions = [
        dict(row)
        for row in _rows(
            conn,
            "roadmap_versions",
            "version_id, era_id, bump_kind, notes, source_round_no, released_at",
        )
    ]
    versions.sort(key=lambda row: row["version_id"])
 
    phases = []
    for row in _rows(
        conn,
        "roadmap_phases",
        "phase_id, phase_code, label, parent_phase_id, branch_suggestion, "
        "commit_prefix, target_version, priority_label, parallel_track, "
        "full_spec_url, sort_order, purpose, introduced_in_version_id, "
        "retired_in_version_id",
    ):
        record = dict(row)
        record["parent_phase_code"] = phase_code_by_id.get(record.pop("parent_phase_id"))
        record.pop("phase_id")
        phases.append(record)
    phases.sort(key=lambda row: row["phase_code"])
 
    tracks = [
        dict(row)
        for row in _rows(
            conn,
            "roadmap_tracks",
            "track_code, label, description, sort_order, active",
        )
    ]
    tracks.sort(key=lambda row: row["track_code"])
 
    items = []
    for row in _rows(
        conn,
        "roadmap_items_v3",
        "item_id, track_kind, phase_id, parent_item_id, title, prose, status, "
        "priority, sort_order, target_version, superseded_by_item_id, "
        "successor_item_id, purpose, introduced_in_version_id, closed_in_version_id",
    ):
        record = dict(row)
        record["phase_code"] = phase_code_by_id.get(record.pop("phase_id"))
        items.append(record)
    items.sort(key=lambda row: row["item_id"])
 
    item_tracks = []
    for row in _rows(conn, "roadmap_item_tracks", "item_id, track_id"):
        item_tracks.append(
            {"item_id": row["item_id"], "track_code": track_code_by_id.get(row["track_id"])}
        )
    item_tracks.sort(key=lambda row: (row["item_id"], row["track_code"] or ""))
 
    dependencies = [
        {"from_item_id": row["from_item_id"], "to_item_id": row["to_item_id"], "kind": row["kind"]}
        for row in _rows(
            conn, "roadmap_item_dependencies", "from_item_id, to_item_id, kind"
        )
    ]
    dependencies.sort(key=lambda row: (row["from_item_id"], row["to_item_id"], row["kind"]))
 
    render_sections = []
    for row in _rows(
        conn,
        "roadmap_render_sections",
        "file_kind, section_path, sort_order, query_kind, query_params_json, "
        "required_min_version, active",
    ):
        record = dict(row)
        record["query_params_json"] = _canonical_query_params(
            record["query_params_json"], record["section_path"]
        )
        render_sections.append(record)
    render_sections.sort(key=lambda row: (row["file_kind"], row["section_path"]))
 
    return {
        "hash_preimage_schema_version": HASH_PREIMAGE_SCHEMA_VERSION,
        "roadmap_eras": eras,
        "roadmap_versions": versions,
        "roadmap_phases": phases,
        "roadmap_tracks": tracks,
        "roadmap_items_v3": items,
        "roadmap_item_tracks": item_tracks,
        "roadmap_item_dependencies": dependencies,
        "roadmap_render_sections": render_sections,
    }
 
 
def logical_state_hash(conn: sqlite3.Connection) -> str:
    return _hash_preimage(logical_state_projection(conn))
 
 
# --- §10 failure-envelope hash nullability (AC §8.48) ------------------------
def stage_hashes(
    stage: str,
    *,
    raw: Optional[str] = None,
    source: Optional[str] = None,
    logical: Optional[str] = None,
) -> dict[str, Optional[str]]:
    """Return the three envelope hash fields with the §10 nullability for ``stage``.
 
    scan failure → raw null; parse/local/canonicalisation → source null, raw
    present; cross-record / pre-build → logical null, source present; clean →
    all three present.
    """
    if stage == STAGE_SCAN:
        return {"raw_source_tree_hash": None, "source_hash": None, "logical_state_hash": None}
    if stage == STAGE_PARSE:
        return {"raw_source_tree_hash": raw, "source_hash": None, "logical_state_hash": None}
    if stage in (STAGE_CROSS_RECORD, STAGE_BUILD):
        return {"raw_source_tree_hash": raw, "source_hash": source, "logical_state_hash": None}
    return {"raw_source_tree_hash": raw, "source_hash": source, "logical_state_hash": logical}
 
 
# --- hash-preimage debug surface (§3.2 / §13) -------------------------------
def hash_preimage(
    which: str,
    *,
    root: Optional[Path] = None,
    records: Optional[list[ScannedRecord]] = None,
    loaded: Optional[list[tuple[ScannedRecord, RegistryModel]]] = None,
    conn: Optional[sqlite3.Connection] = None,
) -> dict[str, Any]:
    """Return the canonical preimage for ``which`` projection (debug surface)."""
    if which == "source-tree":
        if records is None:
            raise ValueError("source-tree preimage requires records")
        return source_hash_preimage(records)
    if which == "raw-source-tree":
        if root is None:
            raise ValueError("raw-source-tree preimage requires root")
        return raw_source_tree_preimage(root)
    if which == "semantic":
        if loaded is None:
            raise ValueError("semantic preimage requires loaded records")
        return semantic_preimage(loaded)
    if which == "logical-state":
        if conn is None:
            raise ValueError("logical-state preimage requires a store connection")
        return logical_state_projection(conn)
    raise ValueError(f"unknown preimage kind: {which!r}")