store_read.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/render/roadmap/store_read.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/roadmap/test_store_read.py"
"""
Guarded read layer for the roadmap GeneratedView renderers (Plan 142 P1.6).
 
Purpose
-------
Open the **verified** authored projection (P1.4 reader guard) and project the
INCLUDED §3.2 rows the renderers need — the derived ``current`` pointer
(AC §8.35), the milestone (m_phase) table, per-phase sub-items with ``Done N/M``
(AC §8.50), non-phase item tables, and the era overview — every query routed
through the §8.30 render guard so no runtime-excluded table/column is read.
 
Responsibilities
----------------
- ``open_verified(db_path)`` — open through ``open_roadmap_reader`` (the
  ``REGEN_UNVERIFIED_PROJECTION`` guard).
- ``projection_hashes(conn)`` — the header's ``source_hash`` /
  ``logical_state_hash`` (from the verification stamp; deterministic fallback).
- ``current_phase_code`` / ``selected_roadmap_version`` — the §3.2 derived pointer
  and the rendered version label.
- ``milestone_rows`` / ``sub_items`` / ``non_phase_rows`` / ``era_rows``.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - read lifecycle
    L3 - selection summaries
    L4 - per-row projection tracing
 
Contracts
---------
- Read-only; never mutates the store.
- ``current`` = the first phase in canonical ``sort_order`` whose parent
  ``m_phase`` item is non-terminal (done/deferred/superseded all terminal).
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.regenerate.verified_projection (the reader guard)
- oc_chat.registry.source.hashing (logical_state_hash fallback)
- oc_chat.registry.render.roadmap._common (the render guard + glyphs)
"""
from __future__ import annotations
 
import sqlite3
from dataclasses import dataclass
from pathlib import Path
from typing import Optional
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.regenerate.verified_projection import open_roadmap_reader, stamp_hashes
from oc_chat.registry.source.hashing import logical_state_hash
from oc_chat.registry.render.roadmap._common import TERMINAL_STATUSES, guarded_query
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
#: All-complete sentinel rendered when ``current`` is null (§3.2 byte-pinned).
ALL_COMPLETE_LINE = "**Current:** _all phases complete_"
 
 
@dataclass(frozen=True)
class Milestone:
    """A structural M-phase row (the parent ``m_phase`` ``roadmap_items_v3`` row)."""
 
    phase_code: str
    title: str
    purpose: str
    prose: Optional[str]
    priority: Optional[str]
    status: str
    sort_order: int
    target_version: Optional[str]
    successor: Optional[str]
    superseded_by: Optional[str]
    done: int
    total: int
    depends_on: tuple[str, ...]
 
 
@dataclass(frozen=True)
class Item:
    """A non-phase or sub-phase ``roadmap_items_v3`` row (INCLUDED columns only)."""
 
    item_id: str
    track_kind: str
    title: str
    purpose: str
    prose: Optional[str]
    status: str
    priority: Optional[str]
    sort_order: int
    successor: Optional[str]
    superseded_by: Optional[str]
    blocked_by: tuple[str, ...] = ()
 
 
def open_verified(db_path: Path) -> sqlite3.Connection:
    """Open ``db_path`` through the P1.4 verified-projection reader guard."""
    return open_roadmap_reader(Path(db_path))
 
 
def projection_hashes(conn: sqlite3.Connection) -> tuple[str, str]:
    """Return ``(source_hash, logical_state_hash)`` for the generated header.
 
    Read from the durable verification stamp (written at regenerate step-7). An
    ``authored_projection`` store MUST carry a ``passed`` stamp — a missing or
    non-``passed`` stamp fails closed with ``REGEN_UNVERIFIED_PROJECTION``
    (Plan 142i R2(a), via ``stamp_hashes``). For any other store (a legacy live
    ingest store — the F7 pass-through), fall back to the recomputed logical hash
    for both — keeps ``render --check`` deterministic.
    """
    hashes = stamp_hashes(conn)
    if hashes is not None:
        return hashes
    computed = logical_state_hash(conn)
    return computed, computed
 
 
def selected_roadmap_version(conn: sqlite3.Connection) -> Optional[str]:
    """Return the ``current`` roadmap version id, else the highest, else None."""
    row = guarded_query(
        conn,
        # OCT-LINT: disable=no_hardcoded_secrets reason="Read-only SQL projection strings are schema constants, not secrets"
        "SELECT version_id FROM roadmap_versions "
        "ORDER BY status = 'current' DESC, version_id DESC LIMIT 1",
    ).fetchone()
    return None if row is None else str(row[0])
 
 
def _child_counts(conn: sqlite3.Connection) -> dict[str, tuple[int, int]]:
    """Return ``{phase_code: (done, total)}`` over each phase's child items."""
    counts: dict[str, tuple[int, int]] = {}
    for row in guarded_query(
        conn,
        # OCT-LINT: disable=no_hardcoded_secrets reason="Read-only SQL projection strings are schema constants, not secrets"
        "SELECT parent_item_id AS parent, "
        "SUM(CASE WHEN status = 'done' THEN 1 ELSE 0 END) AS done, COUNT(*) AS total "
        "FROM roadmap_items_v3 WHERE parent_item_id IS NOT NULL "
        "GROUP BY parent_item_id",
    ).fetchall():
        counts[str(row["parent"])] = (int(row["done"]), int(row["total"]))
    return counts
 
 
#: Dependency kinds that constitute an active "🔒 Blocked by" relation (a soft
#: ``enables`` edge is not a block); ``part_of`` is the parent relation, not a row.
_BLOCKING_KINDS = ("blocks", "gates")
 
 
def _dependencies(conn: sqlite3.Connection) -> dict[str, tuple[str, ...]]:
    """Return ``{dependent_item_id: (predecessor_id, ...)}`` for ``blocks``/``gates``
    edges only (§3.2 direction) — the predecessors a 🔒 blocker annotation names."""
    deps: dict[str, list[str]] = {}
    for row in guarded_query(
        conn,
        "SELECT from_item_id, to_item_id, kind FROM roadmap_item_dependencies "
        "ORDER BY to_item_id, from_item_id",
    ).fetchall():
        if str(row["kind"]) not in _BLOCKING_KINDS:
            continue
        deps.setdefault(str(row["to_item_id"]), []).append(str(row["from_item_id"]))
    return {key: tuple(value) for key, value in deps.items()}
 
 
def item_titles(conn: sqlite3.Connection) -> dict[str, str]:
    """Return ``{item_id: title}`` for the 🔒 ``Blocked by <id> — <title>`` lookup."""
    conn.row_factory = sqlite3.Row
    return {
        str(row["item_id"]): str(row["title"])
        for row in guarded_query(conn, "SELECT item_id, title FROM roadmap_items_v3").fetchall()
    }
 
 
def item_statuses(conn: sqlite3.Connection) -> dict[str, str]:
    """Return ``{item_id: status}`` for the blocker filter — only an *incomplete*
    (non-terminal) predecessor constitutes an active 🔒 block; a satisfied
    (done/deferred/superseded) predecessor is not shown."""
    conn.row_factory = sqlite3.Row
    return {
        str(row["item_id"]): str(row["status"])
        for row in guarded_query(conn, "SELECT item_id, status FROM roadmap_items_v3").fetchall()
    }
 
 
def milestone_rows(conn: sqlite3.Connection) -> list[Milestone]:
    """Return the structural M-phase rows in canonical ``sort_order`` (AC §8.50)."""
    conn.row_factory = sqlite3.Row
    counts = _child_counts(conn)
    deps = _dependencies(conn)
    rows = guarded_query(
        conn,
        # OCT-LINT: disable=no_hardcoded_secrets reason="Read-only SQL projection strings are schema constants, not secrets"
        "SELECT i.item_id, i.title, i.purpose, i.prose, i.priority, i.status, "
        "i.sort_order, i.target_version, i.successor_item_id, i.superseded_by_item_id "
        "FROM roadmap_items_v3 i "
        "JOIN roadmap_phases p ON p.phase_code = i.item_id "
        "WHERE i.track_kind = 'm_phase' AND i.parent_item_id IS NULL "
        "ORDER BY i.sort_order, i.item_id",
    ).fetchall()
    milestones: list[Milestone] = []
    for row in rows:
        done, total = counts.get(str(row["item_id"]), (0, 0))
        milestones.append(
            Milestone(
                phase_code=str(row["item_id"]),
                title=str(row["title"]),
                purpose=str(row["purpose"]),
                prose=row["prose"],
                priority=row["priority"],
                status=str(row["status"]),
                sort_order=int(row["sort_order"]),
                target_version=row["target_version"],
                successor=row["successor_item_id"],
                superseded_by=row["superseded_by_item_id"],
                done=done,
                total=total,
                depends_on=deps.get(str(row["item_id"]), ()),
            )
        )
    return milestones
 
 
def current_phase_code(conn: sqlite3.Connection) -> Optional[str]:
    """Derive the ``current`` pointer (§3.2 / AC §8.35).
 
    The first phase, in canonical ``sort_order``, whose parent ``m_phase`` item is
    non-terminal (``done``/``deferred``/``superseded`` are all terminal). Null
    when every phase is terminal (all-complete behaviour).
    """
    func = "current_phase_code"
    for milestone in milestone_rows(conn):
        if milestone.status not in TERMINAL_STATUSES:
            _dbg("OC_CHAT.MAIN", func, milestone.phase_code, 3)
            return milestone.phase_code
    return None
 
 
def sub_items(conn: sqlite3.Connection, phase_code: str) -> list[Item]:
    """Return a phase's child items (sub-phases + promoted) in ``sort_order``,
    each carrying its ``blocks``/``gates`` predecessors (for the 🔒 annotation)."""
    deps = _dependencies(conn)
    conn.row_factory = sqlite3.Row
    rows = guarded_query(
        conn,
        # OCT-LINT: disable=no_hardcoded_secrets reason="Read-only SQL projection strings are schema constants, not secrets"
        "SELECT item_id, track_kind, title, purpose, prose, status, priority, "
        "sort_order, successor_item_id, superseded_by_item_id "
        "FROM roadmap_items_v3 WHERE parent_item_id = ? "
        "ORDER BY sort_order, item_id",
        (phase_code,),
    ).fetchall()
    return [_to_item(row, deps.get(str(row["item_id"]), ())) for row in rows]
 
 
def non_phase_rows(conn: sqlite3.Connection, track_kinds: tuple[str, ...]) -> list[Item]:
    """Return non-phase item rows for the given ``track_kinds`` in ``sort_order``."""
    if not track_kinds:
        return []
    conn.row_factory = sqlite3.Row
    placeholders = ",".join("?" for _ in track_kinds)
    rows = guarded_query(
        conn,
        # OCT-LINT: disable=no_hardcoded_secrets reason="Read-only SQL projection strings are schema constants, not secrets"
        "SELECT item_id, track_kind, title, purpose, prose, status, priority, "
        "sort_order, successor_item_id, superseded_by_item_id "
        # OCT-LINT: disable=no_hardcoded_secrets reason="Read-only SQL projection strings are schema constants, not secrets"
        f"FROM roadmap_items_v3 WHERE track_kind IN ({placeholders}) "
        "AND phase_id IS NULL "
        "ORDER BY track_kind, sort_order, item_id",
        tuple(track_kinds),
    ).fetchall()
    return [_to_item(row) for row in rows]
 
 
def era_rows(conn: sqlite3.Connection) -> list[sqlite3.Row]:
    """Return era rows in ``sort_order`` for the era-overview section."""
    conn.row_factory = sqlite3.Row
    return guarded_query(
        conn,
        "SELECT era_id, era_label, major_bump_reason, started_at, ended_at, sort_order "
        "FROM roadmap_eras ORDER BY sort_order, era_id",
    ).fetchall()
 
 
def _to_item(row: sqlite3.Row, blocked_by: tuple[str, ...] = ()) -> Item:
    return Item(
        item_id=str(row["item_id"]),
        track_kind=str(row["track_kind"]),
        title=str(row["title"]),
        purpose=str(row["purpose"]),
        prose=row["prose"],
        status=str(row["status"]),
        priority=row["priority"],
        sort_order=int(row["sort_order"]),
        successor=row["successor_item_id"],
        superseded_by=row["superseded_by_item_id"],
        blocked_by=tuple(blocked_by),
    )
 
 
__all__ = [
    "ALL_COMPLETE_LINE",
    "Milestone",
    "Item",
    "open_verified",
    "projection_hashes",
    "selected_roadmap_version",
    "milestone_rows",
    "current_phase_code",
    "sub_items",
    "non_phase_rows",
    "era_rows",
    "item_titles",
    "item_statuses",
]