invariant_matrix.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/invariant_matrix.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_invariant_matrix.py"
# OCT-LINT: disable=no_hardcoded_secrets reason="SQL keywords, table/column names and SRC_* codes are schema constants"
"""
The V0_3(.1) invariant-preservation matrix generator (Plan 142 P1.5, §5.3.1).
 
Purpose
-------
Close the ``regenerate``-bypass class by construction for the DDL half and
machine-gate it against drift for the application-layer half (Registry Source &
Storage Spec V0.22 §5.3.1 / AC §8.49). ``regenerate`` builds the temp DB by
**direct insert**, not through ``execute_domain_command``, so every
application-layer-only invariant is silently bypassed on rebuild and must be
independently handled. This module builds the **two machine-extracted
inventories** that anchor the matrix:
 
1. A **DDL-constraint inventory** parsed from the consolidated
   ``REGISTRY_SCHEMA_V0_3_1.sql`` (the post-amendment baseline) plus the
   ``registry_meta.store_scope`` constraint emitted by
   ``v0_3_2_roadmap_source_fields.migration_sql()`` — one stable-ID record for
   every enforceable constraint (CHECK / unique index / ``UNIQUE`` / PK / FK /
   ``NOT NULL`` / trigger); plain non-unique performance indexes are excluded.
2. An **application-layer ``@invariant`` inventory** extracted from
   ``@invariant: <id>`` markers in the ``REGISTRY_ROADMAP_SCHEMA_V0_3_1``
   amendment doc (HTML-comment + SQL-fence forms) and the rebuild enforcement
   code (code-comment form).
 
Each inventory ID is classified exactly once (``D``/``R``/``X``/``NA``) by the
authored :data:`CLASSIFICATION` map; :func:`generate_matrix` joins the parsed
inventories to that map and applies the D-safety predicate, so the test suite
(``test_invariant_matrix.py``) can assert correctness, the two mutation gates,
the ``{code} ⊆ {matrix}`` drift gate, and the reverse-inclusion R gate.
 
Why ``.1.sql`` (not ``V0_3.sql`` + ``.1.sql``)
----------------------------------------------
AC §8.49 names "``REGISTRY_SCHEMA_V0_3.sql`` + the consolidated
``REGISTRY_SCHEMA_V0_3_1.sql``". The consolidated ``.1.sql`` is, by its own
header, the fresh-create state **after** applying ``V0_3.sql`` then the
``v0_3_2`` migration — i.e. a superset of every ``V0_3`` roadmap constraint with
the amendments folded in. Parsing both would double-count (duplicate stable
IDs); parsing the consolidated baseline alone covers every ``V0_3`` constraint
with no loss. The one constraint the consolidated ``.1.sql`` omits is the base
``registry_meta.store_scope`` (a base-capability ALTER, not part of the roadmap
snapshot — umbrella §12 wave-1 follow-up), which is recovered from
``migration_sql()``.
 
Responsibilities
----------------
- :func:`parse_ddl_inventory` — parse a ``.sql`` file into ``DdlConstraint`` records.
- :func:`extract_markers` — extract ``@invariant: <id>`` markers from md/sql/py text.
- :func:`generate_matrix` — join inventories to :data:`CLASSIFICATION`; return a
  :class:`Matrix` with the first-class derive-verify rows + the validity hashes.
- :func:`build_audit_payload` — the §9.1 audit ``.json`` payload (counts + hashes).
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - generation lifecycle
    L3 - inventory + classification joins
    L4 - per-constraint parse tracing
 
Contracts
---------
- Pure / read-only: parses files + the authored classification; never mutates them.
- Deterministic: inventories + matrix rows are order-stable (sorted by stable ID).
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.migrations.v0_3_2_roadmap_source_fields (store_scope DDL)
"""
from __future__ import annotations
 
import hashlib
import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.migrations import v0_3_2_roadmap_source_fields
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
# --- Repository anchors ------------------------------------------------------
_ORCH = Path(__file__).resolve().parents[1] / "docs" / "architecture" / "orchestration"
#: The consolidated post-amendment DDL baseline (the matrix's ``.1.sql`` input).
SCHEMA_V0_3_1_SQL = _ORCH / "REGISTRY_SCHEMA_V0_3_1.sql"
#: The live aggregating amendment doc carrying the prose ``@invariant`` markers.
V0_3_1_DOC = _ORCH / "REGISTRY_ROADMAP_SCHEMA_V0_3_1.md"
 
_REGEN = Path(__file__).resolve().parent / "regenerate"
#: The rebuild enforcement code the matrix tags + drift-gates ({code} ⊆ {matrix}).
#: These are the §5.3 source-validation / derive-verify sites; the live
#: ``operations.py`` path is deliberately OUT of the matrix (§5.3.1 Codex V18 F3:
#: the matrix is scoped to ``store_scope='authored_projection'`` — live commands
#: are governed by the ``store_scope`` guard, not this matrix).
ENFORCEMENT_CODE_FILES: tuple[Path, ...] = (
    _REGEN / "cross_record.py",
    _REGEN / "derive_verify.py",
    _REGEN / "build.py",
    _REGEN / "pipeline.py",
)
 
# --- Table populations (§5.3.1 D-safety predicate) --------------------------
#: Tables the authored rebuild writes (a constraint on one of these is exercised).
SOURCE_POPULATED_TABLES: frozenset[str] = frozenset({
    "roadmap_eras", "roadmap_versions", "roadmap_phases", "roadmap_tracks",
    "roadmap_items_v3", "roadmap_item_tracks", "roadmap_item_dependencies",
    "roadmap_render_sections",
})
#: Tables the authored rebuild never writes → every constraint on them is NA.
NEVER_WRITTEN_TABLES: frozenset[str] = frozenset({
    "roadmap_item_events", "roadmap_plan_links",
})
#: Runtime-only FK targets — a FK into one fails D-safety predicate (i) → X/NA.
RUNTIME_TABLES: frozenset[str] = frozenset({
    "agents", "rounds", "plans", "operation_log",
})
 
#: Classification tokens (§5.3.1).
VALID_CLASSES: frozenset[str] = frozenset({"D", "R", "X", "NA"})
#: The three admissible D-safety witnesses (§5.3.1 / AC §8.49).
WITNESS_TYPES: frozenset[str] = frozenset({
    "source_validation_rule", "projection_fill_rule", "derive_verify_rule",
})
 
 
# ===========================================================================
# Data model
# ===========================================================================
@dataclass(frozen=True)
class DdlConstraint:
    """One enforceable DDL constraint with a stable ID (the CI anchor)."""
 
    stable_id: str
    table: str
    kind: str  # pk | unique_col | unique_index | fk | notnull | check | trigger
    detail: str  # human-readable expression / column / target
    target_table: Optional[str] = None  # for fk
    partial: bool = False  # for a partial unique index
 
 
@dataclass(frozen=True)
class Marker:
    """One extracted ``@invariant: <id>`` marker and where it was found."""
 
    invariant_id: str
    source: str  # the file (rel) the marker was extracted from
    form: str  # html | sql | code
 
 
@dataclass(frozen=True)
class MatrixRow:
    """An authored classification of one inventory ID (the human-readable view,
    machine-precise). The ``test_v0_3_invariant_matrix_complete`` predicate runs
    over these."""
 
    classification: str  # D | R | X | NA
    handling: str  # one-line rationale
    # D rows name the witness that makes the violation unreachable:
    witness_type: Optional[str] = None  # source_validation_rule | projection_fill_rule | derive_verify_rule
    witness_id: Optional[str] = None
    # R rows bind the source-validation assertion that rejects the violation:
    source_validation_rule: Optional[str] = None  # the @invariant id on the assertion
    acceptance_criterion: Optional[str] = None  # the V0.22 §8 AC #
    failure_row: Optional[str] = None  # the §10 SRC_*/REGEN_* code + test
    negative_fixture: Optional[str] = None  # the violating-source test
    # X rows name the V0_3_1 ratification rationale:
    ratification: Optional[str] = None
    # derive_verify D rows are first-class in the generated inventory (§5.3.1):
    derive_rule_id: Optional[str] = None
    source_fields: tuple[str, ...] = ()
    stored_columns: tuple[str, ...] = ()
    verify_phase: Optional[str] = None
    read_barrier: Optional[str] = None
    positive_fixtures: tuple[str, ...] = ()
    error_code: Optional[str] = None
 
 
@dataclass(frozen=True)
class Matrix:
    """The generated matrix: the joined inventories + their classifications."""
 
    ddl: tuple[DdlConstraint, ...]
    doc_markers: tuple[Marker, ...]
    code_markers: tuple[Marker, ...]
    rows: dict[str, MatrixRow]  # stable_id -> MatrixRow (every inventory id)
    constraint_inventory_hash: str
    app_invariant_inventory_hash: str
 
 
# ===========================================================================
# DDL inventory parser
# ===========================================================================
_CREATE_TABLE_RE = re.compile(
    r"CREATE TABLE(?: IF NOT EXISTS)?\s+(\w+)\s*\‍((.*?)\n\‍)\s*;",
    re.DOTALL | re.IGNORECASE,
)
_CREATE_UNIQUE_INDEX_RE = re.compile(
    r"CREATE\s+UNIQUE\s+INDEX(?: IF NOT EXISTS)?\s+(\w+)\s+ON\s+(\w+)\s*\‍(([^)]*)\‍)"
    r"(?:\s+WHERE\s+(.+?))?\s*;",
    re.DOTALL | re.IGNORECASE,
)
_CREATE_TRIGGER_RE = re.compile(
    r"CREATE\s+TRIGGER(?: IF NOT EXISTS)?\s+(\w+)\s+(BEFORE|AFTER)\s+(\w+)\s+ON\s+(\w+)",
    re.IGNORECASE,
)
_ALTER_ADD_COLUMN_RE = re.compile(
    r"ALTER TABLE\s+(\w+)\s+ADD COLUMN\s+(.+?);",
    re.DOTALL | re.IGNORECASE,
)
 
 
def _slug(text: str) -> str:
    """Deterministic readable slug from a SQL expression (stable across runs)."""
    lowered = text.lower()
    lowered = lowered.replace("!=", "_ne_").replace("=", "_eq_")
    cleaned = re.sub(r"[^a-z0-9]+", "_", lowered).strip("_")
    return re.sub(r"_+", "_", cleaned)
 
 
def _split_table_body(body: str) -> list[str]:
    """Split a CREATE TABLE body into top-level comma-separated clauses (paren-aware)."""
    clauses: list[str] = []
    depth = 0
    current: list[str] = []
    for char in body:
        if char == "(":
            depth += 1
        elif char == ")":
            depth -= 1
        if char == "," and depth == 0:
            clauses.append("".join(current).strip())
            current = []
        else:
            current.append(char)
    tail = "".join(current).strip()
    if tail:
        clauses.append(tail)
    return clauses
 
 
_SQL_RESERVED_CLAUSE_HEADS = ("CHECK", "PRIMARY", "UNIQUE", "FOREIGN", "CONSTRAINT")
 
 
def _parse_column_clause(table: str, clause: str) -> list[DdlConstraint]:
    """Parse a single column-definition clause into its constraints."""
    out: list[DdlConstraint] = []
    name = clause.split()[0]
    upper = clause.upper()
 
    if "PRIMARY KEY" in upper:
        out.append(DdlConstraint(f"ddl:{table}:pk:{name}", table, "pk", name))
    # A column-level UNIQUE (not the surrogate PK).
    if re.search(r"\bUNIQUE\b", upper) and "PRIMARY KEY" not in upper:
        out.append(DdlConstraint(f"ddl:{table}:unique_col:{name}", table, "unique_col", name))
    # NOT NULL (PRIMARY KEY columns are implicitly NOT NULL; do not double-count).
    if "NOT NULL" in upper and "PRIMARY KEY" not in upper:
        out.append(DdlConstraint(f"ddl:{table}:notnull:{name}", table, "notnull", name))
    # Column-level REFERENCES (foreign key).
    ref = re.search(r"REFERENCES\s+(\w+)", clause, re.IGNORECASE)
    if ref:
        target = ref.group(1)
        out.append(DdlConstraint(
            f"ddl:{table}:fk:{name}->{target}", table, "fk", f"{name} -> {target}",
            target_table=target,
        ))
    # Column-level CHECK (e.g. enum / range).
    col_check = re.search(r"CHECK\s*\‍((.+)\‍)", clause, re.IGNORECASE | re.DOTALL)
    if col_check:
        expr = " ".join(col_check.group(1).split())
        out.append(DdlConstraint(
            f"ddl:{table}:check:{name}", table, "check", expr,
        ))
    return out
 
 
def _parse_table_constraint_clause(table: str, clause: str) -> list[DdlConstraint]:
    """Parse a table-level clause (composite PK, table CHECK, table FK)."""
    out: list[DdlConstraint] = []
    upper = clause.upper()
    if upper.startswith("PRIMARY KEY"):
        cols = re.search(r"PRIMARY KEY\s*\‍((.+)\‍)", clause, re.IGNORECASE)
        col_text = _slug(cols.group(1)) if cols else "composite"
        out.append(DdlConstraint(f"ddl:{table}:pk:{col_text}", table, "pk", clause.strip()))
    elif upper.startswith("CHECK"):
        expr_m = re.search(r"CHECK\s*\‍((.+)\‍)\s*$", clause, re.IGNORECASE | re.DOTALL)
        expr = " ".join((expr_m.group(1) if expr_m else clause).split())
        out.append(DdlConstraint(
            f"ddl:{table}:check:{_slug(expr)}", table, "check", expr,
        ))
    elif upper.startswith("UNIQUE"):
        cols = re.search(r"UNIQUE\s*\‍((.+)\‍)", clause, re.IGNORECASE)
        col_text = _slug(cols.group(1)) if cols else "cols"
        out.append(DdlConstraint(
            f"ddl:{table}:unique_cols:{col_text}", table, "unique_index", clause.strip(),
        ))
    elif upper.startswith("FOREIGN KEY"):
        fk = re.search(r"FOREIGN KEY\s*\‍((.+?)\‍)\s*REFERENCES\s+(\w+)", clause, re.IGNORECASE)
        if fk:
            col_text = _slug(fk.group(1))
            target = fk.group(2)
            out.append(DdlConstraint(
                f"ddl:{table}:fk:{col_text}->{target}", table, "fk",
                clause.strip(), target_table=target,
            ))
    return out
 
 
def _parse_create_tables(sql: str) -> list[DdlConstraint]:
    out: list[DdlConstraint] = []
    for match in _CREATE_TABLE_RE.finditer(sql):
        table = match.group(1)
        if table.endswith("_new"):
            continue  # the table-rebuild scratch table — not a published constraint
        body = match.group(2)
        for clause in _split_table_body(body):
            if not clause:
                continue
            head = clause.split("(", 1)[0].strip().upper().split()
            head_word = head[0] if head else ""
            if head_word in _SQL_RESERVED_CLAUSE_HEADS:
                out.extend(_parse_table_constraint_clause(table, clause))
            else:
                out.extend(_parse_column_clause(table, clause))
    return out
 
 
def _parse_indexes_and_triggers(sql: str) -> list[DdlConstraint]:
    out: list[DdlConstraint] = []
    for match in _CREATE_UNIQUE_INDEX_RE.finditer(sql):
        index_name, table, cols, where = match.groups()
        if table.endswith("_new"):
            continue
        partial = where is not None
        out.append(DdlConstraint(
            f"ddl:{table}:unique_index:{index_name}", table, "unique_index",
            f"{index_name}({' '.join(cols.split())})"
            + (f" WHERE {' '.join(where.split())}" if partial else ""),
            partial=partial,
        ))
    for match in _CREATE_TRIGGER_RE.finditer(sql):
        trigger_name, _timing, _event, table = match.groups()
        if table.endswith("_new"):
            continue
        out.append(DdlConstraint(
            f"ddl:{table}:trigger:{trigger_name}", table, "trigger", trigger_name,
        ))
    return out
 
 
def _parse_store_scope() -> list[DdlConstraint]:
    """Recover ``registry_meta.store_scope`` from the v0_3_2 migration ALTER (the
    consolidated ``.1.sql`` omits the base ``registry_meta`` table — umbrella §12)."""
    out: list[DdlConstraint] = []
    sql = v0_3_2_roadmap_source_fields.migration_sql()
    for match in _ALTER_ADD_COLUMN_RE.finditer(sql):
        table, coldef = match.groups()
        if table != "registry_meta":
            continue
        # The store_scope ALTER is a single column def; reuse the column parser.
        out.extend(_parse_column_clause(table, " ".join(coldef.split())))
    return out
 
 
def parse_ddl_inventory(
    sql_path: Path = SCHEMA_V0_3_1_SQL,
    *,
    include_store_scope: bool = True,
) -> list[DdlConstraint]:
    """Parse ``sql_path`` (+ the ``registry_meta.store_scope`` migration ALTER)
    into a deterministic, de-duplicated DDL-constraint inventory."""
    sql = sql_path.read_text(encoding="utf-8")
    constraints = _parse_create_tables(sql) + _parse_indexes_and_triggers(sql)
    if include_store_scope:
        constraints += _parse_store_scope()
    # De-dup by stable id (a repeated constraint is the same invariant) + sort.
    seen: dict[str, DdlConstraint] = {}
    for constraint in constraints:
        seen.setdefault(constraint.stable_id, constraint)
    return sorted(seen.values(), key=lambda c: c.stable_id)
 
 
# ===========================================================================
# @invariant marker extraction
# ===========================================================================
# The id charset includes ``>`` so a marker may name a raw DDL stable id
# (e.g. ``ddl:roadmap_versions:fk:source_round_no->rounds``).
_HTML_MARKER_RE = re.compile(r"<!--\s*@invariant:\s*([A-Za-z0-9._:>-]+)\s*-->")
_SQL_MARKER_RE = re.compile(r"--\s*@invariant:\s*([A-Za-z0-9._:>-]+)")
# OCT-LINT: disable=no_hardcoded_secrets reason="SQL keywords, table/column names and SRC_* codes are schema constants"
_CODE_MARKER_RE = re.compile(r"#\s*@invariant:\s*([A-Za-z0-9._:>-]+)")
 
 
def extract_markers(text: str, *, form: str, source: str) -> list[Marker]:
    """Extract ``@invariant: <id>`` markers from ``text`` in the given ``form``
    (``html`` = Markdown prose, ``sql`` = SQL fence, ``code`` = Python comment)."""
    if form == "html":
        ids = _HTML_MARKER_RE.findall(text)
        # SQL-fence form inside a Markdown doc is also accepted (§5.3.1 / AC §8.49).
        ids += _SQL_MARKER_RE.findall(text)
    elif form == "sql":
        ids = _SQL_MARKER_RE.findall(text)
    elif form == "code":
        ids = _CODE_MARKER_RE.findall(text)
    else:  # pragma: no cover - guarded by callers
        raise ValueError(f"unknown marker form {form!r}")
    return [Marker(invariant_id=mid, source=source, form=form) for mid in ids]
 
 
def extract_doc_markers(doc_path: Path = V0_3_1_DOC) -> list[Marker]:
    """Extract the prose ``@invariant`` markers from the V0_3_1 amendment doc."""
    return extract_markers(
        doc_path.read_text(encoding="utf-8"), form="html", source=doc_path.name
    )
 
 
def extract_code_markers(files: tuple[Path, ...] = ENFORCEMENT_CODE_FILES) -> list[Marker]:
    """Extract ``@invariant`` markers from the rebuild enforcement code."""
    markers: list[Marker] = []
    for path in files:
        if not path.exists():
            continue
        markers.extend(extract_markers(
            path.read_text(encoding="utf-8"), form="code", source=path.name
        ))
    return markers
 
 
# ===========================================================================
# Hashing (the audit's validity key — §9.1 item 1)
# ===========================================================================
def _hash_lines(lines: list[str]) -> str:
    digest = hashlib.sha256()
    for line in sorted(lines):
        digest.update(line.encode("utf-8"))
        digest.update(b"\n")
    return digest.hexdigest()
 
 
def constraint_inventory_hash(ddl: list[DdlConstraint]) -> str:
    """A stable hash over the DDL inventory (stable IDs) — the stamp + audit key."""
    return _hash_lines([c.stable_id for c in ddl])
 
 
def app_invariant_inventory_hash(markers: list[Marker]) -> str:
    """A stable hash over the extracted app-layer ``@invariant`` ids."""
    return _hash_lines(sorted({m.invariant_id for m in markers}))
 
 
# ===========================================================================
# The witness registry (the D-safety predicate's mechanical half — §5.3.1)
# ===========================================================================
# Each D row names a witness that makes its violation unreachable from valid
# authored source. The registry records the witness type + where it is enforced;
# ``test_v0_3_invariant_matrix_complete`` asserts every D's witness is registered
# (exists) and that derive-verify witnesses cover their stored columns.
#
# source_validation_rule witnesses live in the (P1.1) pydantic contracts /
# (P1.3) loader and are NAMED here (P1.5 does not tag P1.1/P1.3 surfaces);
# projection_fill_rule witnesses live in the (P1.4) builder; the single
# derive_verify_rule witness is tagged in ``derive_verify.py``.
WITNESS_REGISTRY: dict[str, tuple[str, str]] = {
    # --- source_validation_rule (pydantic / loader — unreachable from valid source) ---
    "sv.status_enum": ("source_validation_rule",
                       "option_c_registry_contracts.base.Status Literal (§7 enum)"),
    "sv.priority_enum": ("source_validation_rule",
                         "option_c_registry_contracts.base.Priority Literal (§7 enum)"),
    "sv.track_kind_enum": ("source_validation_rule",
                           # OCT-LINT: disable=no_hardcoded_secrets reason="SQL keywords, table/column names and SRC_* codes are schema constants"
                           "base.NonPhaseTrackKind + builder 'm_phase' (§7 enum)"),
    "sv.dep_kind_enum": ("source_validation_rule",
                         "base.DepKind 3-token Literal ⊂ the 4-token DDL CHECK"),
    "sv.lifecycle": ("source_validation_rule",
                     "base.check_item_lifecycle (self-ref, superseded⇒superseded_by, "
                     "successor⇒deferred)"),
    "sv.stem_eq_id": ("source_validation_rule",
                      "canonical.load_record / pipeline._load_promoted stem==id "
                      "(filename-keyed natural id — a duplicate needs two same-named files)"),
    "sv.required_fields": ("source_validation_rule",
                           "pydantic required strict types (RequiredLine / Line300 / "
                           "EntityId / IsoDate / int) — a NULL/missing authored value is rejected on load"),
    # --- projection_fill_rule (P1.4 builder — byte-stable fill, never authored) ---
    "pf.timestamps": ("projection_fill_rule",
                      # OCT-LINT: disable=no_hardcoded_secrets reason="SQL keywords, table/column names and SRC_* codes are schema constants"
                      "build.SOURCE_EPOCH fill for created_at/updated_at/last_updated_at (EXCLUDED)"),
    "pf.created_by_slot": ("projection_fill_rule",
                           # OCT-LINT: disable=no_hardcoded_secrets reason="SQL keywords, table/column names and SRC_* codes are schema constants"
                           "build.SOURCE_SLOT 'source' placeholder NOT-NULL fill (FK half is X)"),
    "pf.store_scope": ("projection_fill_rule",
                       "store_scope.stamp_authored_projection writes 'authored_projection'"),
    "pf.parallel_track": ("projection_fill_rule", "build sets parallel_track=0 (no source field)"),
    "pf.parent_item_id_derived": ("projection_fill_rule",
                                  "parent_item_id set only from sub_phase nesting / promoted dir "
                                  "(never authored) — a self-parent is unreachable"),
    "pf.phase_parent_null": ("projection_fill_rule",
                             "build always sets parent_phase_id=NULL (v1 has no nested phases)"),
    "pf.surrogate_pk": ("projection_fill_rule",
                        # OCT-LINT: disable=no_hardcoded_secrets reason="SQL keywords, table/column names and SRC_* codes are schema constants"
                        "SQLite AUTOINCREMENT assigns the surrogate phase_id/track_id/id"),
    "pf.item_phase_id": ("projection_fill_rule",
                         # OCT-LINT: disable=no_hardcoded_secrets reason="SQL keywords, table/column names and SRC_* codes are schema constants"
                         "phase fan-out sets phase_id for every m_phase row; non-phase items get NULL "
                         "(no source phase_id field) — C17/C18 unreachable"),
    "pf.insert_or_ignore": ("projection_fill_rule",
                            "build INSERT OR IGNORE dedups the composite membership/dependency PK"),
    "pf.track_id_resolved": ("projection_fill_rule",
                             # OCT-LINT: disable=no_hardcoded_secrets reason="SQL keywords, table/column names and SRC_* codes are schema constants"
                             "build resolves track_code→track_id (SRC_BAD_TRACK_CODES if unknown); "
                             "the surrogate is never NULL"),
    "pf.render_sections": ("projection_fill_rule",
                           # OCT-LINT: disable=no_hardcoded_secrets reason="SQL keywords, table/column names and SRC_* codes are schema constants"
                           "render_sections.seed_render_sections deterministic builder-fill from "
                           "the constant manifest (no §3.1 source entity, never authored); every "
                           "file_kind/query_kind/active value satisfies its CHECK by construction "
                           "and required_min_version is NULL → the FK never fires (P1.6 seam)"),
    "sv.track_active": ("source_validation_rule",
                        "base.Track.active bool (default True) → always a valid 0/1"),
    # --- derive_verify_rule (P1.4 derive_verify — tagged @invariant) ---
    "inv.version_lineage_derived": ("derive_verify_rule",
                                    "derive_verify.derive_versions + verify_versions (step-7 publish-gate)"),
}
 
 
# ===========================================================================
# The authored classification (the human-readable §5.3.1 view, machine-precise)
# ===========================================================================
def _D(witness_id: str, handling: str, **kw: object) -> MatrixRow:
    wtype = WITNESS_REGISTRY[witness_id][0]
    return MatrixRow(
        classification="D", handling=handling,
        witness_type=wtype, witness_id=witness_id, **kw,  # type: ignore[arg-type]
    )
 
 
def _R(rule: str, ac: str, code: str, neg_fixture: str, handling: str) -> MatrixRow:
    return MatrixRow(
        classification="R", handling=handling, source_validation_rule=rule,
        acceptance_criterion=ac, failure_row=code, negative_fixture=neg_fixture,
    )
 
 
def _X(ratification: str, handling: str) -> MatrixRow:
    return MatrixRow(classification="X", handling=handling, ratification=ratification)
 
 
def _NA(handling: str) -> MatrixRow:
    return MatrixRow(classification="NA", handling=handling)
 
 
# The single first-class derive-verify row (§5.3.1 — recorded with all fields so
# "not R" never means "not in the app-layer audit surface").
_VERSION_LINEAGE_ROW = MatrixRow(
    classification="D",
    # OCT-LINT: disable=no_hardcoded_secrets reason="SQL keywords, table/column names and SRC_* codes are schema constants"
    handling="version status / lineage (C3–C6) / one-current-per-era / bump_kind are "
             "DERIVED from the ordered changelog (numeric-per-field) then recomputed-and-"
             "verified at step-7 before any use; a violating value is unreachable in the built store",
    witness_type="derive_verify_rule",
    witness_id="inv.version_lineage_derived",
    derive_rule_id="derive_verify.derive_versions",
    source_fields=("version", "era", "date", "round", "bump_override"),
    stored_columns=("status", "parent_version_id", "supersedes_version_id",
                    "superseded_by_version_id", "bump_kind"),
    verify_phase="step-7",
    # OCT-LINT: disable=no_hardcoded_secrets reason="SQL keywords, table/column names and SRC_* codes are schema constants"
    read_barrier="verified_projection.assert_verified_projection (REGEN_UNVERIFIED_PROJECTION)",
    positive_fixtures=("test_version_ordering_numeric_across_major_boundary",
                       "test_version_selection_skip_superseded",
                       "test_version_selection_per_era_cardinality"),
    error_code="REGEN_BUILD_FAILED",
)
 
_DV = "inv.version_lineage_derived"
 
#: stable_id -> MatrixRow for the SOURCE-POPULATED tables + registry_meta (the
#: rows the predicate does not assign mechanically; never-written + render_v1
#: tables are filled programmatically by :func:`_build_classification`).
_EXPLICIT: dict[str, MatrixRow] = {
    # --- registry_meta (store_scope base capability, §9.3 / AC §8.57) -------
    "ddl:registry_meta:check:store_scope": _D("pf.store_scope",
        "builder stamps a valid scope token; author cannot write registry_meta"),
    "ddl:registry_meta:notnull:store_scope": _D("pf.store_scope",
        "builder always stamps store_scope before any runtime open"),
    # --- roadmap_eras -------------------------------------------------------
    "ddl:roadmap_eras:notnull:era_label": _D("sv.required_fields", "Era.era_label RequiredLine"),
    "ddl:roadmap_eras:notnull:sort_order": _D("sv.required_fields", "Era.order required int"),
    "ddl:roadmap_eras:notnull:started_at": _D("sv.required_fields", "Era.started_at IsoDate"),
    "ddl:roadmap_eras:pk:era_id": _D("sv.stem_eq_id",
        "era_id is the filename stem (stem==id enforced); a duplicate needs two same-named files"),
    "ddl:roadmap_eras:unique_index:idx_roadmap_eras_sort": _R(
        "inv.era_track_order_unique", "8.56", "SRC_DUPLICATE_ORDER",
        "test_duplicate_era_order_rejected", "two eras with the same order — author-reachable"),
    # --- roadmap_versions ---------------------------------------------------
    "ddl:roadmap_versions:check:bump_kind": _D(_DV, "bump_kind derived (verified step-7)"),
    "ddl:roadmap_versions:check:not_status_eq_superseded_and_superseded_by_version_id_is_null": _D(
        _DV, "C6: lineage derived — superseded⇒superseded_by unreachable in built store"),
    "ddl:roadmap_versions:check:status": _D(_DV, "status derived (draft/current/superseded)"),
    "ddl:roadmap_versions:check:version_id_ne_parent_version_id": _D(_DV, "C3: lineage derived (no authored field)"),
    "ddl:roadmap_versions:check:version_id_ne_superseded_by_version_id": _D(_DV, "C5: lineage derived"),
    "ddl:roadmap_versions:check:version_id_ne_supersedes_version_id": _D(_DV, "C4: lineage derived"),
    "ddl:roadmap_versions:fk:era_id->roadmap_eras": _R(
        "inv.intra_source_fk_resolves", "8.32", "REGEN_BUILD_FAILED",
        "test_dangling_version_era_fails", "version.era may reference a missing era — author-reachable"),
    "ddl:roadmap_versions:fk:parent_version_id->roadmap_versions": _D(_DV, "lineage derived within the version set"),
    "ddl:roadmap_versions:fk:source_round_no->rounds": _X("V0_3_1 §9.2",
        "runtime-reference relaxation (Option A): authored provenance, FK not enforced on the authored rebuild"),
    "ddl:roadmap_versions:fk:superseded_by_version_id->roadmap_versions": _D(_DV, "lineage derived"),
    "ddl:roadmap_versions:fk:supersedes_version_id->roadmap_versions": _D(_DV, "lineage derived"),
    "ddl:roadmap_versions:notnull:bump_kind": _D(_DV, "bump_kind derived (always set)"),
    "ddl:roadmap_versions:notnull:era_id": _D("sv.required_fields", "Version.era required EntityId"),
    "ddl:roadmap_versions:notnull:last_updated_at": _D("pf.timestamps", "SOURCE_EPOCH fill (EXCLUDED)"),
    "ddl:roadmap_versions:notnull:released_at": _D("sv.required_fields", "Version.date IsoDate (released_at)"),
    "ddl:roadmap_versions:notnull:status": _D(_DV, "status derived (always set)"),
    "ddl:roadmap_versions:pk:version_id": _R(
        "inv.version_metadata_agreement", "8.38", "SRC_VERSION_METADATA_CONFLICT",
        "test_version_metadata_conflict", "a roadmap+product pair shares version_id — merged to one row, agreement-checked"),
    "ddl:roadmap_versions:unique_index:idx_roadmap_versions_one_current_per_era": _D(
        _DV, "one-current-per-era derived + verified step-7 (AC §8.26)"),
    # --- roadmap_phases -----------------------------------------------------
    "ddl:roadmap_phases:check:parallel_track": _D("pf.parallel_track", "builder sets 0 (no source field)"),
    "ddl:roadmap_phases:check:phase_id_ne_parent_phase_id": _D("pf.phase_parent_null",
        "C9: parent_phase_id always NULL on the rebuild"),
    "ddl:roadmap_phases:check:priority_label": _D("sv.priority_enum", "Phase.priority Priority enum"),
    "ddl:roadmap_phases:fk:introduced_in_version_id->roadmap_versions": _R(
        "inv.intra_source_fk_resolves", "8.32", "REGEN_BUILD_FAILED",
        "test_dangling_phase_version_pointer_fails", "introduced_in_version may dangle — author-reachable"),
    "ddl:roadmap_phases:fk:parent_phase_id->roadmap_phases": _D("pf.phase_parent_null",
        "parent_phase_id always NULL (v1 has no nested phases) — resolves trivially"),
    "ddl:roadmap_phases:fk:retired_in_version_id->roadmap_versions": _R(
        "inv.intra_source_fk_resolves", "8.32", "REGEN_BUILD_FAILED",
        "test_dangling_phase_version_pointer_fails", "retired_in_version may dangle — author-reachable"),
    "ddl:roadmap_phases:notnull:created_at": _D("pf.timestamps", "SOURCE_EPOCH fill"),
    "ddl:roadmap_phases:notnull:label": _D("sv.required_fields", "Phase.title RequiredLine (label)"),
    "ddl:roadmap_phases:notnull:parallel_track": _D("pf.parallel_track", "builder sets 0"),
    "ddl:roadmap_phases:notnull:phase_code": _D("sv.required_fields", "Phase.id EntityId (phase_code)"),
    "ddl:roadmap_phases:notnull:purpose": _D("sv.required_fields", "Phase.purpose Line300 (non-empty)"),
    "ddl:roadmap_phases:notnull:sort_order": _D("sv.required_fields", "Phase.order required int"),
    "ddl:roadmap_phases:notnull:updated_at": _D("pf.timestamps", "SOURCE_EPOCH fill"),
    # OCT-LINT: disable=no_hardcoded_secrets reason="SQL keywords, table/column names and SRC_* codes are schema constants"
    "ddl:roadmap_phases:pk:phase_id": _D("pf.surrogate_pk", "AUTOINCREMENT surrogate (source-populated)"),
    "ddl:roadmap_phases:unique_col:phase_code": _R(
        "inv.phase_track_code_unique", "8.56", "SRC_DUPLICATE_CODE",
        "test_duplicate_phase_code_rejected", "two phases with the same code — author-reachable"),
    # --- roadmap_tracks -----------------------------------------------------
    "ddl:roadmap_tracks:check:active": _D("sv.track_active", "Track.active bool → 0/1"),
    "ddl:roadmap_tracks:notnull:active": _D("sv.track_active", "Track.active bool (default True)"),
    "ddl:roadmap_tracks:notnull:label": _D("sv.required_fields", "Track.label RequiredLine"),
    "ddl:roadmap_tracks:notnull:sort_order": _D("sv.required_fields", "Track.order required int"),
    "ddl:roadmap_tracks:notnull:track_code": _D("sv.required_fields", "Track.track_code EntityId"),
    "ddl:roadmap_tracks:pk:track_id": _D("pf.surrogate_pk", "AUTOINCREMENT surrogate"),
    "ddl:roadmap_tracks:unique_col:track_code": _R(
        "inv.phase_track_code_unique", "8.56", "SRC_DUPLICATE_CODE",
        "test_duplicate_track_code_rejected", "two tracks with the same code — author-reachable"),
    "ddl:roadmap_tracks:unique_index:idx_roadmap_tracks_sort": _R(
        "inv.era_track_order_unique", "8.56", "SRC_DUPLICATE_ORDER",
        "test_duplicate_track_order_rejected", "two tracks with the same order — author-reachable"),
    # --- roadmap_items_v3 ---------------------------------------------------
    "ddl:roadmap_items_v3:check:item_id_ne_parent_item_id": _D("pf.parent_item_id_derived",
        "C14: parent_item_id is builder-derived; a self-parent is unreachable"),
    "ddl:roadmap_items_v3:check:item_id_ne_superseded_by_item_id": _D("sv.lifecycle",
        "C15: check_item_lifecycle rejects superseded_by==id"),
    "ddl:roadmap_items_v3:check:priority": _D("sv.priority_enum", "RoadmapItem.priority Priority enum (nullable)"),
    "ddl:roadmap_items_v3:check:status": _D("sv.status_enum", "Status enum"),
    "ddl:roadmap_items_v3:check:status_ne_superseded_or_superseded_by_item_id_is_not_null": _D(
        "sv.lifecycle", "C16: check_item_lifecycle requires superseded_by when status=superseded"),
    "ddl:roadmap_items_v3:check:track_kind": _D("sv.track_kind_enum", "track_kind enum"),
    "ddl:roadmap_items_v3:check:track_kind_ne_m_phase_or_phase_id_is_not_null": _D("pf.item_phase_id",
        "C17: phase fan-out always sets phase_id for m_phase rows"),
    "ddl:roadmap_items_v3:check:track_kind_not_in_backlog_vision_or_phase_id_is_null": _D("pf.item_phase_id",
        "C18: non-phase items have no source phase_id field → builder sets NULL"),
    "ddl:roadmap_items_v3:fk:closed_in_version_id->roadmap_versions": _R(
        "inv.intra_source_fk_resolves", "8.32", "REGEN_BUILD_FAILED",
        "test_dangling_item_version_pointer_fails", "closed_in_version may dangle — author-reachable"),
    "ddl:roadmap_items_v3:fk:closed_round_no->rounds": _X("V0_3_1 §9.2",
        # OCT-LINT: disable=no_hardcoded_secrets reason="SQL keywords, table/column names and SRC_* codes are schema constants"
        "runtime-reference relaxation (Option A): always NULL on a pure source rebuild, EXCLUDED from the hash"),
    "ddl:roadmap_items_v3:fk:introduced_in_version_id->roadmap_versions": _R(
        "inv.intra_source_fk_resolves", "8.32", "REGEN_BUILD_FAILED",
        "test_dangling_item_version_pointer_fails", "introduced_in_version may dangle — author-reachable"),
    "ddl:roadmap_items_v3:fk:parent_item_id->roadmap_items_v3": _D("pf.parent_item_id_derived",
        "parent_item_id derived from nesting/promoted dir → always resolves to a built item"),
    "ddl:roadmap_items_v3:fk:phase_id->roadmap_phases": _D("pf.item_phase_id",
        "phase_id derived from the fan-out → always resolves"),
    "ddl:roadmap_items_v3:fk:successor_item_id->roadmap_items_v3": _R(
        "inv.intra_source_fk_resolves", "8.32", "REGEN_BUILD_FAILED",
        "test_dangling_successor_pointer_fails", "successor is authored and may dangle — author-reachable"),
    "ddl:roadmap_items_v3:fk:superseded_by_item_id->roadmap_items_v3": _R(
        "inv.intra_source_fk_resolves", "8.32", "REGEN_BUILD_FAILED",
        "test_dangling_superseded_by_pointer_fails", "superseded_by is authored and may dangle — author-reachable"),
    "ddl:roadmap_items_v3:notnull:created_at": _D("pf.timestamps", "SOURCE_EPOCH fill"),
    "ddl:roadmap_items_v3:notnull:purpose": _D("sv.required_fields", "purpose Line300 (non-empty)"),
    "ddl:roadmap_items_v3:notnull:sort_order": _D("sv.required_fields", "order required int"),
    "ddl:roadmap_items_v3:notnull:status": _D("sv.status_enum", "status enum (always set)"),
    "ddl:roadmap_items_v3:notnull:title": _D("sv.required_fields", "title RequiredLine"),
    "ddl:roadmap_items_v3:notnull:track_kind": _D("sv.track_kind_enum", "track_kind enum (always set)"),
    "ddl:roadmap_items_v3:notnull:updated_at": _D("pf.timestamps", "SOURCE_EPOCH fill"),
    "ddl:roadmap_items_v3:pk:item_id": _R(
        "inv.item_id_unique", "8.39", "SRC_ITEM_ID_COLLISION",
        "test_item_id_union_collision_case_insensitive", "an id may collide across file ids ∪ nested sub_phase ids ∪ promoted"),
    # --- roadmap_item_tracks ------------------------------------------------
    "ddl:roadmap_item_tracks:fk:item_id->roadmap_items_v3": _D("pf.parent_item_id_derived",
        "the membership owner is the item being built — always present"),
    "ddl:roadmap_item_tracks:fk:track_id->roadmap_tracks": _R(
        "inv.track_codes_resolve", "8.31", "SRC_BAD_TRACK_CODES",
        "test_unknown_track_code_rejected", "track_codes may reference an unknown track — author-reachable"),
    "ddl:roadmap_item_tracks:notnull:created_at": _D("pf.timestamps", "SOURCE_EPOCH fill"),
    "ddl:roadmap_item_tracks:notnull:item_id": _D("sv.required_fields", "membership owner id"),
    "ddl:roadmap_item_tracks:notnull:track_id": _D("pf.track_id_resolved", "resolved track_code→track_id"),
    "ddl:roadmap_item_tracks:pk:item_id_track_id": _D("pf.insert_or_ignore", "INSERT OR IGNORE dedups duplicates"),
    # --- roadmap_item_dependencies ------------------------------------------
    "ddl:roadmap_item_dependencies:check:from_item_id_ne_to_item_id": _R(
        "inv.dependency_acyclic", "8.47", "SRC_DEPENDENCY_CYCLE",
        "test_dependency_self_edge_rejected", "C20: a self-edge is a degenerate cycle — author-reachable"),
    "ddl:roadmap_item_dependencies:check:kind": _D("sv.dep_kind_enum",
        "the 3 authorable kinds are a subset of the 4-token DDL CHECK; the builder never writes part_of"),
    "ddl:roadmap_item_dependencies:fk:created_by_slot->agents": _X("V0_3_1 §9.2",
        # OCT-LINT: disable=no_hardcoded_secrets reason="SQL keywords, table/column names and SRC_* codes are schema constants"
        "runtime-reference relaxation (Option A): FK-to-agents half (the NOT-NULL half is D via 'source' placeholder)"),
    "ddl:roadmap_item_dependencies:fk:from_item_id->roadmap_items_v3": _R(
        "inv.depends_on_resolves", "8.3", "SRC_DANGLING_DEPENDS_ON",
        "test_dangling_depends_on_rejected", "depends_on predecessor may be unknown — author-reachable"),
    "ddl:roadmap_item_dependencies:fk:to_item_id->roadmap_items_v3": _R(
        "inv.depends_on_resolves", "8.3", "SRC_DANGLING_DEPENDS_ON",
        "test_dangling_depends_on_rejected", "the dependent is the owning item; resolution shares the depends_on check"),
    "ddl:roadmap_item_dependencies:notnull:created_at": _D("pf.timestamps", "SOURCE_EPOCH fill"),
    "ddl:roadmap_item_dependencies:notnull:created_by_slot": _D("pf.created_by_slot", "'source' placeholder fill"),
    "ddl:roadmap_item_dependencies:notnull:from_item_id": _D("sv.required_fields", "depends_on edge id (EntityId)"),
    "ddl:roadmap_item_dependencies:notnull:kind": _D("sv.dep_kind_enum", "DepKind defaults to 'blocks' (always set)"),
    "ddl:roadmap_item_dependencies:notnull:to_item_id": _D("sv.required_fields", "the owning item id"),
    "ddl:roadmap_item_dependencies:pk:from_item_id_to_item_id_kind": _D("pf.insert_or_ignore",
        "INSERT OR IGNORE dedups a repeated (from,to,kind) edge"),
}
 
# App-layer invariants that have no DDL constraint OR carry the X/derive
# semantics — the §9.1 audit checklist + the doc/code markers reference these.
_APP_LAYER: dict[str, MatrixRow] = {
    # The first-class derive-verify row.
    _DV: _VERSION_LINEAGE_ROW,
    # The bounded-suspension inv-(b) — app-layer only, no DDL form (dormant in v1).
    "inv.b_closed_version_required": _X("V0_3_1 §9.1",
        "inv-(b) done/deferred⇒closed_in_version: a deliberate post-cutover tightening, "
        "grandfathered by the terminal-at-cutover seed ledger; v1 ships lenient (dormant)"),
    # The cross-record R invariants (the step-6 / build / pipeline assertions).
    "inv.item_id_unique": _R("inv.item_id_unique", "8.39", "SRC_ITEM_ID_COLLISION",
        "test_item_id_union_collision_case_insensitive", "id unique across file ids ∪ nested ∪ promoted (step-6)"),
    "inv.depends_on_resolves": _R("inv.depends_on_resolves", "8.3", "SRC_DANGLING_DEPENDS_ON",
        "test_dangling_depends_on_rejected", "every depends_on predecessor resolves (step-6)"),
    "inv.dependency_acyclic": _R("inv.dependency_acyclic", "8.47", "SRC_DEPENDENCY_CYCLE",
        "test_dependency_cycle_rejected_from_source", "kind-unscoped dependency acyclicity (step-6, OQ-R-07)"),
    "inv.successor_acyclic": _R("inv.successor_acyclic", "8.42", "SRC_SUCCESSOR_CYCLE",
        "test_successor_cycle_rejected", "successor-chain acyclicity (step-6)"),
    "inv.done_parent_terminal_children": _R("inv.done_parent_terminal_children", "8.50",
        "SRC_DONE_PARENT_LIVE_CHILD", "test_done_parent_requires_terminal_children",
        "a done parent has only terminal children (step-6, OQ-R-07)"),
    "inv.era_track_order_unique": _R("inv.era_track_order_unique", "8.56", "SRC_DUPLICATE_ORDER",
        "test_duplicate_era_order_rejected", "era/track order uniqueness pre-check (step-6)"),
    "inv.phase_track_code_unique": _R("inv.phase_track_code_unique", "8.56", "SRC_DUPLICATE_CODE",
        "test_duplicate_phase_code_rejected", "phase_code/track_code uniqueness pre-check (step-6)"),
    "inv.version_metadata_agreement": _R("inv.version_metadata_agreement", "8.38",
        "SRC_VERSION_METADATA_CONFLICT", "test_version_metadata_conflict",
        "same-version roadmap+product metadata agreement → one row (step-6)"),
    "inv.track_codes_resolve": _R("inv.track_codes_resolve", "8.31", "SRC_BAD_TRACK_CODES",
        "test_unknown_track_code_rejected", "every track_code resolves to a track (build)"),
    "inv.intra_source_fk_resolves": _R("inv.intra_source_fk_resolves", "8.32", "REGEN_BUILD_FAILED",
        "test_dangling_version_era_fails",
        "intra-source FK resolution backstop (post-build foreign_key_check; Option A tolerates runtime FKs)"),
}
 
 
def _build_classification() -> dict[str, MatrixRow]:
    """Assemble the authoritative classification: explicit source-populated +
    app-layer rows, plus programmatic NA for the never-written + v1-empty tables.
 
    Built once from the canonical ``.1.sql`` inventory so the mutation tests
    (which parse a *fixture* ``.sql``) see injected ids absent from this map.
    """
    classification: dict[str, MatrixRow] = dict(_EXPLICIT)
    classification.update(_APP_LAYER)
    for constraint in parse_ddl_inventory():
        if constraint.stable_id in classification:
            continue
        if constraint.table in NEVER_WRITTEN_TABLES:
            classification[constraint.stable_id] = _NA(
                f"{constraint.table} is never written on the authored rebuild "
                "(events/plan-links are runtime-only; AC §8.23) — never exercised")
        elif constraint.table == "roadmap_render_sections":
            classification[constraint.stable_id] = _D("pf.render_sections",
                # OCT-LINT: disable=no_hardcoded_secrets reason="SQL keywords, table/column names and SRC_* codes are schema constants"
                "roadmap_render_sections has no §3.1 source entity but is SEEDED deterministically "
                "on every rebuild by P1.6 render_sections.seed_render_sections (the P1.4↔P1.6 seam, "
                "wired in build.build_roadmap); the constant manifest satisfies every CHECK/NOT-NULL "
                "by construction and seeds required_min_version NULL → the FK never fires")
    return classification
 
 
#: stable_id -> MatrixRow. Every DDL inventory id AND every app-layer @invariant
#: id appears here exactly once (test_v0_3_invariant_matrix_complete).
CLASSIFICATION: dict[str, MatrixRow] = _build_classification()
 
 
def generate_matrix(
    sql_path: Path = SCHEMA_V0_3_1_SQL,
    doc_path: Path = V0_3_1_DOC,
    code_files: tuple[Path, ...] = ENFORCEMENT_CODE_FILES,
) -> Matrix:
    """Join the two machine-extracted inventories to :data:`CLASSIFICATION`."""
    func = "generate_matrix"
    ddl = parse_ddl_inventory(sql_path)
    doc_markers = extract_doc_markers(doc_path)
    code_markers = extract_code_markers(code_files)
    all_markers = doc_markers + code_markers
    rows = {
        cid: CLASSIFICATION[cid]
        for cid in ({c.stable_id for c in ddl} | {m.invariant_id for m in all_markers})
        if cid in CLASSIFICATION
    }
    _dbg("OC_CHAT.MAIN", func,
         f"ddl={len(ddl)} doc_markers={len(doc_markers)} code_markers={len(code_markers)}", 3)
    return Matrix(
        ddl=tuple(ddl),
        doc_markers=tuple(doc_markers),
        code_markers=tuple(code_markers),
        rows=rows,
        constraint_inventory_hash=constraint_inventory_hash(ddl),
        app_invariant_inventory_hash=app_invariant_inventory_hash(all_markers),
    )
 
 
# ===========================================================================
# The §9.1 application-layer checklist + audit payload
# ===========================================================================
def verify_matrix(matrix: Matrix) -> list[str]:
    """Run the §5.3.1 correctness predicate over ``matrix``; return a list of
    violation strings (empty == the matrix is complete + correct).
 
    Asserts: (a) every DDL-inventory id + every extracted ``@invariant`` id is
    classified **exactly once**; (b) every ``D`` satisfies the D-safety predicate
    (a registered witness; a ``derive_verify_rule`` names stored columns + a
    positive fixture per branch); (c) no catch-all (every row carries a specific
    handling); (d) every ``R`` names a source-validation rule + AC + §10 code +
    negative fixture; (e) every ``X`` names a ``V0_3_1`` ratification; (f) every
    ``NA`` names why the rebuild never exercises it.
    """
    violations: list[str] = []
    classified = matrix.rows
    inventory_ids = {c.stable_id for c in matrix.ddl}
    inventory_ids |= {m.invariant_id for m in matrix.doc_markers}
    inventory_ids |= {m.invariant_id for m in matrix.code_markers}
    for cid in sorted(inventory_ids):
        if cid not in classified:
            violations.append(f"unclassified inventory id (no matrix row): {cid}")
    for cid, row in sorted(classified.items()):
        if row.classification not in VALID_CLASSES:
            violations.append(f"{cid}: invalid class {row.classification!r}")
        if not row.handling.strip():
            violations.append(f"{cid}: catch-all — no handling rationale")
        if row.classification == "D":
            if row.witness_id not in WITNESS_REGISTRY:
                violations.append(f"{cid}: D names no registered witness ({row.witness_id!r})")
            elif row.witness_type != WITNESS_REGISTRY[row.witness_id][0]:
                violations.append(f"{cid}: D witness_type mismatch vs registry")
            if row.witness_type == "derive_verify_rule":
                # The first-class derive-verify witness row (§5.3.1) carries the
                # branch-coverage evidence; every constraint D-via-derive-verify
                # references it. Validate that witness row exists + is complete.
                witness_row = classified.get(row.witness_id or "")
                if witness_row is None:
                    violations.append(f"{cid}: derive_verify witness {row.witness_id!r} has no first-class row")
                else:
                    if not witness_row.stored_columns:
                        violations.append(f"{row.witness_id}: derive_verify_rule names no stored_columns")
                    if not witness_row.positive_fixtures:
                        violations.append(
                            f"{row.witness_id}: derive_verify_rule names no positive_fixture (branch coverage)")
        elif row.classification == "R":
            if not row.source_validation_rule:
                violations.append(f"{cid}: R names no source_validation_rule")
            if not row.acceptance_criterion:
                violations.append(f"{cid}: R names no acceptance criterion")
            if not row.failure_row:
                violations.append(f"{cid}: R names no §10 failure code")
            if not row.negative_fixture:
                violations.append(f"{cid}: R names no negative fixture")
        elif row.classification == "X":
            if not row.ratification:
                violations.append(f"{cid}: X names no V0_3_1 ratification")
    return violations
 
 
def code_markers_not_in_matrix(matrix: Matrix) -> list[str]:
    """{code-enforced @invariant ids} ⊄ {matrix ids} — the drift gate (AC §8.49)."""
    return sorted(
        m.invariant_id for m in matrix.code_markers if m.invariant_id not in matrix.rows
    )
 
 
def unimplemented_r_rows(matrix: Matrix) -> list[str]:
    """Matrix ``R`` ids whose ``source_validation_rule`` has no enforcement-code
    ``@invariant`` marker — the reverse-inclusion R-completeness gate (AC §8.49)."""
    code_ids = {m.invariant_id for m in matrix.code_markers}
    return sorted(
        cid for cid, row in matrix.rows.items()
        if row.classification == "R" and row.source_validation_rule not in code_ids
    )
 
 
def app_layer_invariant_ids(classification: Optional[dict[str, MatrixRow]] = None) -> set[str]:
    """The §9.1 app-layer-invariant checklist denominator: every invariant the
    rebuild bypasses unless independently handled — the ``inv.*`` rows (R +
    derive-verify) plus the ``X`` conscious relaxations. The pure-DDL ``D``/``NA``
    constraints are by-construction and are NOT app-layer invariants (§5.3.1)."""
    rows = classification if classification is not None else CLASSIFICATION
    return {
        cid for cid, row in rows.items()
        if cid.startswith("inv.") or row.classification == "X"
    }
 
 
def build_audit_payload(matrix: Optional[Matrix] = None) -> dict[str, object]:
    """The machine-readable §9.1 audit ``.json`` payload: the frozen checklist,
    the generated-inventory counts, and the audit's own validity-key hashes
    (the Accepted-doc + enforcement-code marker hashes + the generated-inventory
    hash — NOT the roadmap source_hash, V0.19 Codex V16 F4)."""
    matrix = matrix or generate_matrix()
    counts: dict[str, int] = {"D": 0, "R": 0, "X": 0, "NA": 0}
    for row in matrix.rows.values():
        counts[row.classification] += 1
    checklist = sorted(app_layer_invariant_ids(matrix.rows))
    return {
        "spec_version": "REGISTRY_SOURCE_AND_STORAGE_SPEC_V0_22",
        "generated_by": "oc_chat/registry/invariant_matrix.py",
        "validity_key": {
            "v0_3_1_doc": V0_3_1_DOC.name,
            "doc_marker_inventory_hash": app_invariant_inventory_hash(list(matrix.doc_markers)),
            "code_marker_inventory_hash": app_invariant_inventory_hash(list(matrix.code_markers)),
            "constraint_inventory_hash": matrix.constraint_inventory_hash,
        },
        "inventory_counts": {
            "ddl_constraints": len(matrix.ddl),
            "classified": len(matrix.rows),
            **{f"class_{k}": v for k, v in counts.items()},
            "app_layer_checklist": len(checklist),
        },
        "app_layer_checklist": checklist,
        "signer": {
            "name": None,
            "role": None,
            "date": None,
            "statement": None,
            "independent_of_author": None,
        },
        "freeze_point": "142/P1 implementation-acceptance commit",
    }