verified_projection.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/regenerate/verified_projection.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/regenerate/test_verified_projection.py"
# OCT-LINT: disable=no_hardcoded_secrets reason="Column names + stamp constants are schema constants"
"""
The verified-projection publish-gate: stamp + central reader guard (§5.3.1).
 
Purpose
-------
Write a durable verification STAMP into the temp DB BEFORE the atomic swap (so
the data + stamp become visible atomically), and gate EVERY canonical reader of
an authored projection through one check (Registry Source & Storage Spec V0.22
§5.3.1 — verify-before-USE). A missing/failed stamp on an authored projection is
``REGEN_UNVERIFIED_PROJECTION`` (AC §8.26/49). Live stores do not use this guard
(they use the P1.2 ``store_scope`` guard).
 
Responsibilities
----------------
- ``write_verification_stamp(conn, ...)`` — create + populate the stamp row.
- ``assert_verified_projection(conn)`` — raise on an unverified authored projection.
- ``open_roadmap_reader(db_path)`` — open a guarded read connection.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - stamp/guard lifecycle
    L3 - stamp validity check
    L4 - column probes
 
Contracts
---------
- The stamp lives in ``registry_verification_stamp`` (NOT a roadmap_* table, so
  it never enters the §3.2 logical-state hash).
- The guard is a fast in-DB validity check (status + internal-hash consistency),
  NOT a source-tree recomputation (that is ``render --check`` / ``regenerate``).
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.regenerate.envelope (REGEN_UNVERIFIED_PROJECTION)
"""
from __future__ import annotations
 
import sqlite3
from pathlib import Path
from typing import Optional
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.regenerate.envelope import (
    REGEN_UNVERIFIED_PROJECTION,
    RegenerateError,
)
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
VERIFICATION_SCHEMA_VERSION = 1
STAMP_TABLE = "registry_verification_stamp"
_AUTHORED_PROJECTION = "authored_projection"
_PASSED = "passed"
 
_CREATE_STAMP = f"""
CREATE TABLE IF NOT EXISTS {STAMP_TABLE} (
  id                          INTEGER PRIMARY KEY CHECK(id = 1),
  store_scope                 TEXT NOT NULL,
  source_hash                 TEXT NOT NULL,
  logical_state_hash          TEXT NOT NULL,
  derive_rule_set_hash        TEXT,
  constraint_inventory_hash   TEXT,
  verified_at_step            TEXT NOT NULL,
  verification_schema_version INTEGER NOT NULL,
  verification_status         TEXT NOT NULL
)
"""
 
 
def _store_scope(conn: sqlite3.Connection) -> Optional[str]:
    try:
        row = conn.execute("SELECT store_scope FROM registry_meta WHERE id = 1").fetchone()
    except sqlite3.OperationalError:
        return None
    if row is None:
        return None
    return str(row[0])
 
 
def write_verification_stamp(
    conn: sqlite3.Connection,
    *,
    source_hash: str,
    logical_state_hash: str,
    store_scope: str = _AUTHORED_PROJECTION,
    derive_rule_set_hash: Optional[str] = None,
    constraint_inventory_hash: Optional[str] = None,
    status: str = _PASSED,
) -> None:
    """Write the single-row verification stamp into the temp DB (pre-swap)."""
    conn.execute(_CREATE_STAMP)
    conn.execute(f"DELETE FROM {STAMP_TABLE}")
    conn.execute(
        f"INSERT INTO {STAMP_TABLE}(id, store_scope, source_hash, logical_state_hash, "
        "derive_rule_set_hash, constraint_inventory_hash, verified_at_step, "
        "verification_schema_version, verification_status) VALUES(1,?,?,?,?,?,?,?,?)",
        (store_scope, source_hash, logical_state_hash, derive_rule_set_hash,
         constraint_inventory_hash, "step-7", VERIFICATION_SCHEMA_VERSION, status),
    )
    conn.commit()
 
 
def assert_verified_projection(conn: sqlite3.Connection) -> None:
    """Raise ``REGEN_UNVERIFIED_PROJECTION`` if an authored projection is unverified.
 
    A live store (or pre-v0.3.2 DB) is not gated here. An authored projection
    must carry a ``verification_status='passed'`` stamp.
    """
    if _store_scope(conn) != _AUTHORED_PROJECTION:
        return
    try:
        row = conn.execute(
            f"SELECT verification_status FROM {STAMP_TABLE} WHERE id = 1"
        ).fetchone()
    except sqlite3.OperationalError as error:
        raise RegenerateError(
            "authored projection has no verification stamp (unverified)",
            code=REGEN_UNVERIFIED_PROJECTION,
            hint="run 'regenerate' to build + verify before reading",
        ) from error
    if row is None or str(row[0]) != _PASSED:
        raise RegenerateError(
            "authored projection verification stamp is missing or not 'passed'",
            code=REGEN_UNVERIFIED_PROJECTION,
            hint="run 'regenerate' to build + verify before reading",
        )
 
 
def stamp_hashes(conn: sqlite3.Connection) -> Optional[tuple[str, str]]:
    """Return the stamped ``(source_hash, logical_state_hash)``, fail-closed for
    an authored projection (Plan 142i R2(a), §5.4 [R9] closure).
 
    - An ``authored_projection`` store MUST carry a
      ``verification_status='passed'`` stamp: a missing stamp OR a non-``passed``
      status (N-c) raises ``REGEN_UNVERIFIED_PROJECTION`` — the header's hashes
      are never emitted from an unverified authored projection.
    - Any other store (a legacy ``live_store`` / pre-v0.3.2 DB — the deliberate
      F7 pass-through) returns the stamp values if a stamp exists, else ``None``
      so the caller applies its deterministic fallback (never gated).
 
    This is defence-in-depth for direct / future callers: the render open path
    already raises at open time via ``assert_verified_projection`` for a
    stampless authored projection (G5), so through ``open_verified`` the raise
    never fires — the fail-closed unit test calls this on a raw connection.
    """
    scope = _store_scope(conn)
    try:
        row = conn.execute(
            f"SELECT source_hash, logical_state_hash, verification_status "
            f"FROM {STAMP_TABLE} WHERE id = 1"
        ).fetchone()
    except sqlite3.OperationalError:
        row = None
    if scope == _AUTHORED_PROJECTION:
        if row is None or str(row[2]) != _PASSED:
            raise RegenerateError(
                "authored projection stamp is missing or not 'passed' — "
                "refusing to emit unverified header hashes",
                code=REGEN_UNVERIFIED_PROJECTION,
                hint="run 'regenerate' to build + verify before reading",
            )
        return str(row[0]), str(row[1])
    if row is not None:
        return str(row[0]), str(row[1])
    return None
 
 
def stamp_source_hash_if_authored(conn: sqlite3.Connection) -> Optional[str]:
    """Return the stamped canonical ``source_hash`` IFF this is an
    ``authored_projection`` store carrying a ``passed`` verification stamp;
    else ``None`` (Plan 142i R2(b) stamp-conditional gate).
 
    A stampless legacy live store (or any non-authored / non-``passed`` store)
    returns ``None`` so ``render --check``'s whole-tree ``source_hash`` recompute
    is skipped wholesale — a legacy store carries no canonical-source contract to
    enforce and an unconditional compare would false-drift by construction (H1).
    """
    if _store_scope(conn) != _AUTHORED_PROJECTION:
        return None
    try:
        row = conn.execute(
            f"SELECT source_hash, verification_status FROM {STAMP_TABLE} WHERE id = 1"
        ).fetchone()
    except sqlite3.OperationalError:
        return None
    if row is None or str(row[1]) != _PASSED:
        return None
    return str(row[0])
 
 
def open_roadmap_reader(db_path: Path) -> sqlite3.Connection:
    """Open a read connection through the verified-projection guard."""
    conn = sqlite3.connect(str(db_path))
    conn.row_factory = sqlite3.Row
    try:
        assert_verified_projection(conn)
    except Exception:
        conn.close()
        raise
    return conn