v0_3_2_roadmap_source_fields.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/migrations/v0_3_2_roadmap_source_fields.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_schema_v0_3_2.py"
# OCT-LINT: disable=no_hardcoded_secrets reason="Schema object names are migration constants, not secrets"
"""
Runtime migration for roadmap source fields v0.3.2.
 
Purpose
-------
Apply the additive schema changes that V0.22 §0 [OP-1] confirmed for the
roadmap source-of-truth cutover (Plan 142): ``purpose`` columns on
``roadmap_phases`` and ``roadmap_items_v3``; ``successor_item_id`` on
``roadmap_items_v3``; retirement of the ``roadmap_exec_order``/``roadmap_todo``
``file_kind``s via a SQLite table-rebuild of ``roadmap_render_sections``; and
the base-capability ``registry_meta.store_scope`` column.
 
Spec: ``REGISTRY_ROADMAP_SCHEMA_V0_3_1.md`` (L-a), §1–§6, §9.3.
 
Responsibilities
----------------
- Add ``purpose TEXT NOT NULL DEFAULT ''`` to ``roadmap_phases`` and
  ``roadmap_items_v3`` (§2).
- Add ``successor_item_id TEXT`` self-FK to ``roadmap_items_v3`` (§3).
- Rebuild ``roadmap_render_sections`` with the six-token ``file_kind`` CHECK
  and deactivate retired ``file_kind`` rows (§5).
- Add ``registry_meta.store_scope`` base-capability column (§9.3).
- Bump SQLite ``user_version`` to 32.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - migration lifecycle
    L3 - schema version decisions
    L4 - table definition checks
 
Contracts
---------
- Callers own transaction boundaries.
- The migration is idempotent (guarded by ``is_applied``).
- ``rollback`` reverts bookkeeping only; table shape is backward-compatible.
 
Dependencies
------------
- sqlite3 (migration connection)
- oc_diagnostics (_dbg structured logger)
"""
from __future__ import annotations
 
import sqlite3
 
from oc_diagnostics import _dbg
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
ROADMAP_SOURCE_FIELDS_SCHEMA_VERSION = 32
ROADMAP_SOURCE_FIELDS_SCHEMA_DESCRIPTION = (
    # OCT-LINT: disable=no_hardcoded_secrets reason="Schema object names are migration constants, not secrets"
    "roadmap source fields REGISTRY_ROADMAP_SCHEMA_V0_3_1"
)
 
# Probe strings used by is_applied (§5 / §9.3).
# OCT-LINT: disable=no_hardcoded_secrets reason="Schema object names are migration constants, not secrets"
_ROADMAP_EXTENDED_TOKEN = "'roadmap_extended'"
_STORE_SCOPE_COLUMN = "store_scope"
 
 
def current_user_version(conn: sqlite3.Connection) -> int:
    """Return SQLite's integer schema user version."""
 
    return int(conn.execute("PRAGMA user_version").fetchone()[0])
 
 
def apply(conn: sqlite3.Connection) -> None:
    """Apply the v0.3.2 roadmap source fields migration if needed."""
 
    if is_applied(conn):
        conn.execute(
            f"PRAGMA user_version = {ROADMAP_SOURCE_FIELDS_SCHEMA_VERSION}"
        )
        return
    _run_migration(conn)
    conn.execute(
        f"PRAGMA user_version = {ROADMAP_SOURCE_FIELDS_SCHEMA_VERSION}"
    )
 
 
def rollback(conn: sqlite3.Connection) -> None:
    """Reset user_version for tests; table shape remains backward-compatible.
 
    Mirrors ``v0_3_1_auth_events.rollback``: revert bookkeeping only.
    """
 
    conn.execute("PRAGMA user_version = 31")
    conn.execute(
        "DELETE FROM schema_migrations WHERE version = ?",
        (ROADMAP_SOURCE_FIELDS_SCHEMA_VERSION,),
    )
 
 
def is_applied(conn: sqlite3.Connection) -> bool:
    """Return True when all four migration effects are present (L-a §5 / §9.3).
 
    Probe 1 — ``roadmap_phases.purpose`` AND ``roadmap_items_v3.purpose`` exist.
    Probe 2 — ``roadmap_items_v3.successor_item_id`` exists.
    Probe 3 — ``roadmap_render_sections.file_kind`` CHECK contains ``'roadmap_extended'``.
    Probe 4 — ``registry_meta.store_scope`` exists.
    """
 
    if not _has_column(conn, "roadmap_phases", "purpose"):
        return False
    items_cols = _column_names(conn, "roadmap_items_v3")
    if "purpose" not in items_cols or "successor_item_id" not in items_cols:
        return False
    row = conn.execute(
        # OCT-LINT: disable=no_hardcoded_secrets reason="Schema object names are migration constants, not secrets"
        "SELECT sql FROM sqlite_master"
        " WHERE type = 'table' AND name = 'roadmap_render_sections'"
    ).fetchone()
    if row is None:
        return False
    try:
        sql = row["sql"]
    except (TypeError, KeyError, IndexError):
        sql = row[0]
    if _ROADMAP_EXTENDED_TOKEN not in str(sql):
        return False
    return _has_column(conn, "registry_meta", _STORE_SCOPE_COLUMN)
 
 
def migration_sql() -> str:
    """Return SQL used when building a fresh schema in one executescript call.
 
    This string is appended after ``v0_3_1_auth_events.migration_sql()`` in
    ``_run_schema_script`` inside an outer ``BEGIN IMMEDIATE`` transaction.
    No inner BEGIN/COMMIT — the PRAGMA foreign_keys lines are no-ops inside a
    transaction but take effect when ``apply()`` calls executescript in
    autocommit mode.
    """
 
    return _MIGRATION_SQL + f"\nPRAGMA user_version = {ROADMAP_SOURCE_FIELDS_SCHEMA_VERSION};\n"
 
 
def _run_migration(conn: sqlite3.Connection) -> None:
    conn.executescript(_MIGRATION_SQL)
 
 
def _column_names(conn: sqlite3.Connection, table: str) -> set[str]:
    rows = conn.execute(f"PRAGMA table_info({table})").fetchall()
    result: set[str] = set()
    for row in rows:
        try:
            result.add(str(row["name"]))
        except (TypeError, KeyError, IndexError):
            result.add(str(row[1]))
    return result
 
 
def _has_column(conn: sqlite3.Connection, table: str, column: str) -> bool:
    return column in _column_names(conn, table)
 
 
# ---------------------------------------------------------------------------
# Migration SQL (no transaction wrappers — see migration_sql() docstring).
# ---------------------------------------------------------------------------
 
# OCT-LINT: disable=no_hardcoded_secrets reason="Schema object names are migration constants, not secrets"
_MIGRATION_SQL = """
ALTER TABLE roadmap_phases ADD COLUMN purpose TEXT NOT NULL DEFAULT '';
ALTER TABLE roadmap_items_v3 ADD COLUMN purpose TEXT NOT NULL DEFAULT '';
ALTER TABLE roadmap_items_v3 ADD COLUMN successor_item_id TEXT REFERENCES roadmap_items_v3(item_id);
 
PRAGMA foreign_keys = OFF;
 
CREATE TABLE roadmap_render_sections_new (
  id             INTEGER PRIMARY KEY AUTOINCREMENT,
  file_kind      TEXT    NOT NULL
    CHECK(file_kind IN (
      'roadmap','roadmap_exec_order','roadmap_todo','roadmap_changelog',
      'roadmap_extended','root_changelog'
    )),
  section_path   TEXT    NOT NULL,
  sort_order     INTEGER NOT NULL,
  query_kind     TEXT    NOT NULL
    CHECK(query_kind IN (
      'm_phase_table',
      'phase_section',
      'phase_section_with_subrows',
      'backlog_track_table',
      'git_workflow_table',
      'completed_in_version',
      'recently_completed_window',
      'prose_only'
    )),
  query_params_json    TEXT,
  required_min_version TEXT REFERENCES roadmap_versions(version_id),
  active         INTEGER NOT NULL DEFAULT 1 CHECK(active IN (0,1))
);
 
INSERT INTO roadmap_render_sections_new
  (id, file_kind, section_path, sort_order, query_kind,
   query_params_json, required_min_version, active)
SELECT
  id, file_kind, section_path, sort_order, query_kind,
  query_params_json, required_min_version, active
FROM roadmap_render_sections;
 
DROP TABLE roadmap_render_sections;
 
ALTER TABLE roadmap_render_sections_new RENAME TO roadmap_render_sections;
 
CREATE INDEX idx_roadmap_render_sections_file
  ON roadmap_render_sections(file_kind, sort_order);
 
UPDATE roadmap_render_sections
  SET active = 0
  WHERE file_kind IN ('roadmap_exec_order', 'roadmap_todo');
 
PRAGMA foreign_key_check;
 
PRAGMA foreign_keys = ON;
 
ALTER TABLE registry_meta
  ADD COLUMN store_scope TEXT NOT NULL DEFAULT 'live_store'
  CHECK(store_scope IN ('live_store', 'authored_projection'));
"""