derive_verify.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/regenerate/derive_verify.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/regenerate/test_derive_verify.py"
"""
Version status / lineage / bump_kind derivation + the step-7 derive-verify.
 
Purpose
-------
Derive the *derived* ``roadmap_versions`` columns (``status``, ``parent_/
supersedes_/superseded_by_version_id``, ``bump_kind``, ``one-current-per-era``)
from the ordered changelog — pinned to **numeric-per-field** version ordering
(lexical forbidden, §3.2) — and recompute-and-verify them against the built
store at regenerate step-7 BEFORE the publish/swap/hash (the publish-gate;
AC §8.26/32/37, §5.3.1 ``derive_verify_rule``).
 
Responsibilities
----------------
- ``derive_versions(versions)`` — the deterministic per-era derivation (used by
  the builder to insert AND by the verify to compare).
- ``verify_versions(conn, versions)`` — re-derive + hard-fail on any mismatch.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - derivation lifecycle
    L3 - per-era ordering/selection
    L4 - per-version lineage
 
Contracts
---------
- Ordering is numeric-per-field (major int, then 2-digit minor) — a consistent
  cross-major boundary (v9.99 vs v10.00) is numeric-correct, never lexical.
- Highest non-superseded per era = ``current`` (exactly one per era).
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- option_c_registry_contracts (the version regex constants)
- oc_chat.registry.regenerate.envelope (RegenerateError with REGEN_BUILD_FAILED on a derive-verify mismatch)
"""
from __future__ import annotations
 
import re
import sqlite3
from typing import Any, Optional
 
from oc_diagnostics import _dbg
 
from option_c_registry_contracts import ROADMAP_VERSION_RE
 
from oc_chat.registry.regenerate.envelope import REGEN_BUILD_FAILED, RegenerateError
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
_ROADMAP_VERSION_PAT = re.compile(ROADMAP_VERSION_RE)
 
 
def _parse_version(version_id: str) -> Optional[tuple[int, int]]:
    """Return ``(major, minor)`` for a roadmap-axis ``vN.NN``, else ``None``."""
    if not _ROADMAP_VERSION_PAT.fullmatch(version_id):
        return None  # product SemVer → no roadmap_versions row (§3.2)
    major_str, minor_str = version_id[1:].split(".", 1)
    return int(major_str), int(minor_str)
 
 
def _bump_kind(
    current: tuple[int, int], predecessor: Optional[tuple[int, int]], override: Optional[str]
) -> str:
    if override is not None:
        return "out_of_round_revision"
    if predecessor is None:
        return "major"  # first version of an era (§3.2)
    cmajor, cminor = current
    pmajor, pminor = predecessor
    if cmajor != pmajor:
        return "major"
    if cminor // 10 != pminor // 10:
        return "minor"  # crossed a 10-unit boundary (e.g. v3.09 -> v3.10)
    return "patch"
 
 
def derive_versions(versions: list) -> dict[str, dict[str, Any]]:
    """Derive status/lineage/bump_kind per roadmap-axis version (numeric ordering).
 
    Returns ``{version_id: {status, parent_version_id, supersedes_version_id,
    superseded_by_version_id, bump_kind}}``. Product-SemVer records are excluded
    (they create no ``roadmap_versions`` row, §3.2).
    """
    by_era: dict[str, list] = {}
    override_by_id: dict[str, Optional[str]] = {}
    for version in versions:
        parsed = _parse_version(version.version)
        if parsed is None:
            continue
        by_era.setdefault(version.era, []).append((parsed, version.version))
        override_by_id[version.version] = version.bump_override
 
    derived: dict[str, dict[str, Any]] = {}
    for _era, members in by_era.items():
        members.sort(key=lambda entry: entry[0])  # numeric-per-field (R17-1)
        ordered_ids = [version_id for _parsed, version_id in members]
        for index, (parsed, version_id) in enumerate(members):
            predecessor_id = ordered_ids[index - 1] if index > 0 else None
            successor_id = ordered_ids[index + 1] if index + 1 < len(members) else None
            predecessor_parsed = members[index - 1][0] if index > 0 else None
            is_current = index == len(members) - 1
            derived[version_id] = {
                "status": "current" if is_current else "superseded",
                "parent_version_id": predecessor_id,
                "supersedes_version_id": predecessor_id,
                "superseded_by_version_id": successor_id,
                "bump_kind": _bump_kind(parsed, predecessor_parsed, override_by_id[version_id]),
            }
    return derived
 
 
def dedup_version_records(versions: list) -> list:
    """Collapse records sharing a ``version_id`` into one (a same-version
    roadmap + product changelog pair maps to exactly ONE ``roadmap_versions``
    row — §3.2 / AC §8.38). Required metadata is agreement-checked in the step-6
    cross-record pass; here we coalesce the omit-defers-to-present optionals
    (``round``/``bump_override``/``notes``). Insertion order is preserved.
    """
    by_id: dict[str, Any] = {}
    order: list[str] = []
    for version in versions:
        if version.version not in by_id:
            by_id[version.version] = version
            order.append(version.version)
            continue
        current = by_id[version.version]
        updates = {
            field: getattr(version, field)
            for field in ("round", "bump_override", "notes")
            if getattr(current, field) is None and getattr(version, field) is not None
        }
        if updates:
            by_id[version.version] = current.model_copy(update=updates)
    return [by_id[vid] for vid in order]
 
 
_DERIVED_COLUMNS = (
    "status",
    "parent_version_id",
    "supersedes_version_id",
    "superseded_by_version_id",
    "bump_kind",
)
 
 
def verify_versions(conn: sqlite3.Connection, versions: list) -> None:
    """Re-derive + hard-fail (``REGEN_BUILD_FAILED``) on any stored-vs-derived
    mismatch — the step-7 publish-gate (AC §8.26, ``test_version_status_derive_mismatch``)."""
    func = "verify_versions"
    # @invariant: inv.version_lineage_derived  (§5.3.1 D derive_verify_rule — AC §8.26/§8.32;
    # status/lineage C3–C6 + one-current-per-era + bump_kind derived then verified before USE)
    derived = derive_versions(versions)
    conn.row_factory = sqlite3.Row
    stored = {
        row["version_id"]: row
        for row in conn.execute(
            "SELECT version_id, status, parent_version_id, supersedes_version_id, "
            "superseded_by_version_id, bump_kind FROM roadmap_versions"
        )
    }
    if set(stored) != set(derived):
        raise RegenerateError(
            "derive-verify: stored roadmap_versions set != derived set "
            f"(stored {sorted(stored)}, derived {sorted(derived)})",
            code=REGEN_BUILD_FAILED,
            records=sorted(set(stored) ^ set(derived)),
        )
    for version_id, expected in derived.items():
        row = stored[version_id]
        for column in _DERIVED_COLUMNS:
            if row[column] != expected[column]:
                raise RegenerateError(
                    f"derive-verify: {version_id} {column} stored={row[column]!r} "
                    f"derived={expected[column]!r}",
                    code=REGEN_BUILD_FAILED,
                    records=[version_id],
                    hint="the built store's derived version column does not match source",
                )
    _dbg("OC_CHAT.MAIN", func, f"verified {len(derived)} versions", 3)