store_scope.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/store_scope.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_schema_v0_3_2.py"
"""
Central store-scope guard for the OC Chat registry.
 
Purpose
-------
Enforce the V0.22 §5.3.1 / AC §8.57 contract: every live runtime command
(queue, round, auth) must refuse an ``authored_projection`` store with a
clear diagnostic, and the authored-projection rebuild must stamp the store
before any live open can succeed.
 
Responsibilities
----------------
- Expose ``assert_live_store(conn)`` — raises ``StoreScopeError`` when
  ``registry_meta.store_scope = 'authored_projection'``.
- Expose ``stamp_authored_projection(conn)`` — writes the scope stamp
  before the rebuild path returns a store handle.
- Tolerate pre-v0.3.2 databases (missing ``store_scope`` column) by
  treating them as ``live_store`` (lenient mode, dormant in v1).
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - store-scope lifecycle
    L3 - scope check details
    L4 - column presence probes
 
Contracts
---------
- Callers own connection and transaction lifecycle.
- ``assert_live_store`` is idempotent and read-only; safe to call in any
  transaction mode.
- ``stamp_authored_projection`` must be called before the temp store is
  returned to callers; the stamp is permanent for that DB file.
 
Dependencies
------------
- sqlite3 (the caller's connection and its OperationalError)
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
import sqlite3
 
from oc_diagnostics import _dbg
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
_AUTHORED_PROJECTION = "authored_projection"
_LIVE_STORE = "live_store"
 
 
class StoreScopeError(RuntimeError):
    """Raised when a live command targets an authored-projection store."""
 
    def __init__(self, scope: str = _AUTHORED_PROJECTION) -> None:
        super().__init__(
            f"This registry is an '{scope}' store and cannot execute live commands. "
            "Import a runtime checkpoint to upgrade it to a 'live_store'."
        )
        self.scope = scope
 
 
def assert_live_store(conn: sqlite3.Connection) -> None:
    """Raise ``StoreScopeError`` if the connected store is an authored projection.
 
    Silently returns for pre-v0.3.2 databases (missing ``store_scope`` column)
    and for uninitialized databases (missing ``registry_meta`` row) — both are
    treated as ``live_store`` in v1 (dormant guard per L-a §9.3).
    """
 
    try:
        row = conn.execute(
            "SELECT store_scope FROM registry_meta WHERE id = 1"
        ).fetchone()
    except sqlite3.OperationalError:
        return
    if row is None:
        return
    try:
        scope = str(row["store_scope"])
    except (TypeError, KeyError, IndexError):
        scope = str(row[0])
    if scope == _AUTHORED_PROJECTION:
        raise StoreScopeError(scope)
 
 
def stamp_authored_projection(conn: sqlite3.Connection) -> None:
    """Write ``store_scope = 'authored_projection'`` to ``registry_meta``.
 
    Called by the authored-projection rebuild path before returning a store
    handle so that any subsequent live-command open fails fast.
    Callers own the enclosing transaction.
    """
 
    conn.execute(
        "UPDATE registry_meta SET store_scope = ? WHERE id = 1",
        (_AUTHORED_PROJECTION,),
    )