cross_record.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/regenerate/cross_record.py
 
# OCT-LINT: disable=no_hardcoded_secrets reason="Field names and SRC_* codes are schema constants"
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/regenerate/test_cross_record.py"
"""
Step-6 cross-record validation (Plan 142 P1.4, §5.3 step 6).
 
Purpose
-------
Validate the resolved id index (file ids ∪ nested sub_phase ids ∪ promoted ids)
and every cross-record invariant the direct-insert rebuild bypasses (§5.3.1 R
rows): id-uniqueness, dangling ``depends_on``, kind-unscoped dependency
acyclicity, successor acyclicity, ``done``-parent⇒children-terminal,
era/track ``order``/code uniqueness, same-version changelog metadata
agreement (omit-defers-to-present), and roadmap ``changelog_seq``
presence/uniqueness — each with its durable ``SRC_*`` code.
 
Responsibilities
----------------
- ``validate_cross_record(categorized)`` — run all step-6 checks; raise the
  first failing ``SourceValidationError`` with its §10 code + offending records.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - validation lifecycle
    L3 - per-check details
    L4 - graph walk tracing
 
Contracts
---------
- Pure / read-only over the loaded models; raises before any temp-DB insert.
- Deterministic: checks + offending-record lists are order-stable.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- option_c_registry_contracts (TERMINAL_STATUSES, check_successor_acyclicity)
- oc_chat.registry.changelog_seq (check_changelog_seq for the product changelog seq check)
- oc_chat.registry.regenerate.envelope (the SRC_* codes each failing check raises with)
- oc_chat.registry.source.errors (SourceValidationError raised by the first failing check)
"""
from __future__ import annotations
 
from typing import Any
 
from oc_diagnostics import _dbg
 
from option_c_registry_contracts import (
    TERMINAL_STATUSES,
    DependencyEdge,
    check_successor_acyclicity,
)
 
from oc_chat.registry.regenerate.envelope import (
    SRC_DANGLING_DEPENDS_ON,
    SRC_DEPENDENCY_CYCLE,
    SRC_DONE_PARENT_LIVE_CHILD,
    SRC_DUPLICATE_CODE,
    SRC_DUPLICATE_ORDER,
    SRC_FRAMING_DANGLING_ANCHOR,
    SRC_FRAMING_DUPLICATE_ANCHOR,
    SRC_ITEM_ID_COLLISION,
    SRC_PRODUCT_CHANGELOG_SEQ,
    SRC_ROADMAP_CHANGELOG_SEQ,
    SRC_SUCCESSOR_CYCLE,
    SRC_VERSION_METADATA_CONFLICT,
)
from oc_chat.registry.changelog_seq import check_changelog_seq
from oc_chat.registry.source.errors import SourceValidationError
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
# The store-projected version-metadata fields that must agree on a same-version
# roadmap+product pair (§3.2 / AC §8.38).
_VERSION_AGREE_FIELDS = ("era", "date", "round", "bump_override", "notes")
 
 
def _item_models(categorized: dict[str, list]) -> list[tuple[str, Any, str]]:
    """Return ``(item_id, model, location)`` for every item-bearing record."""
    out: list[tuple[str, Any, str]] = []
    for phase, _body in categorized.get("phases", []):
        out.append((phase.id, phase, f"phases/{phase.id}.md"))
        for sub in phase.sub_phases:
            out.append((sub.id, sub, f"phases/{phase.id}.md#{sub.id}"))
    for item, _body in categorized.get("items", []):
        out.append((item.id, item, f"items/{item.track_kind}/{item.id}.md"))
    for scanned, item, _body in categorized.get("promoted", []):
        out.append((item.id, item, scanned.rel_path))
    return out
 
 
def _check_id_uniqueness(items: list[tuple[str, Any, str]]) -> set[str]:
    """Item ids are unique **case-insensitively** across the union (file ids ∪
    nested sub_phase ids ∪ promoted) — Plan 142j R7 (VC-0007, layer 3 semantic
    identity). A collision (exact OR case-only, §3.2) raises the distinct
    ``SRC_ITEM_ID_COLLISION`` (split out of ``SRC_DUP_OR_STEM_MISMATCH``)."""
    groups: dict[str, list[tuple[str, str]]] = {}  # casefold-id -> [(orig_id, location)]
    for item_id, _model, location in items:
        groups.setdefault(item_id.casefold(), []).append((item_id, location))
    for _key, members in groups.items():
        if len(members) > 1:
            # @invariant: inv.item_id_unique  (§5.3.1 R — AC §8.39 / SRC_ITEM_ID_COLLISION,
            # case-insensitive item-id-union split — Plan 142j R7)
            ids = sorted({mid for mid, _loc in members})
            locations = sorted({loc for _mid, loc in members})
            raise SourceValidationError(
                f"item id collision (case-insensitive) across {ids}: {locations}",
                code=SRC_ITEM_ID_COLLISION,
                records=locations,
                hint="item ids are unique case-insensitively across file ids ∪ nested "
                "sub_phase ids ∪ promoted",
            )
    return {item_id for item_id, _model, _location in items}
 
 
def _check_depends_on_and_cycle(items: list[tuple[str, Any, str]], id_index: set[str]) -> None:
    edges: list[tuple[str, str]] = []  # (from=predecessor, to=dependent)
    for item_id, model, location in items:
        for edge in getattr(model, "depends_on", []) or []:
            predecessor = edge.id if isinstance(edge, DependencyEdge) else edge
            if predecessor not in id_index:
                # @invariant: inv.depends_on_resolves  (§5.3.1 R — AC §8.3 / SRC_DANGLING_DEPENDS_ON)
                raise SourceValidationError(
                    f"{location}: depends_on references unknown id {predecessor!r}",
                    code=SRC_DANGLING_DEPENDS_ON,
                    records=[location],
                )
            edges.append((predecessor, item_id))
    cycle = _find_graph_cycle(edges)
    if cycle is not None:
        # @invariant: inv.dependency_acyclic  (§5.3.1 R — AC §8.47 / SRC_DEPENDENCY_CYCLE, kind-unscoped OQ-R-07)
        raise SourceValidationError(
            "dependency cycle (kind-unscoped): " + " -> ".join([*cycle, cycle[0]]),
            code=SRC_DEPENDENCY_CYCLE,
            records=list(cycle),
        )
 
 
def _find_graph_cycle(edges: list[tuple[str, str]]) -> list[str] | None:
    adjacency: dict[str, list[str]] = {}
    for source, target in edges:
        adjacency.setdefault(source, []).append(target)
    color: dict[str, int] = {}  # 0 visiting, 1 done
    stack: list[str] = []
 
    def _walk(node: str) -> list[str] | None:
        color[node] = 0
        stack.append(node)
        for nxt in adjacency.get(node, []):
            if color.get(nxt) == 0:
                return stack[stack.index(nxt):]
            if color.get(nxt) is None:
                found = _walk(nxt)
                if found is not None:
                    return found
        stack.pop()
        color[node] = 1
        return None
 
    for start in list(adjacency):
        if color.get(start) is None:
            found = _walk(start)
            if found is not None:
                return found
    return None
 
 
def _check_successor_acyclicity(items: list[tuple[str, Any, str]]) -> None:
    holders = [model for _id, model, _loc in items if hasattr(model, "successor")]
    try:
        check_successor_acyclicity(holders)
    except ValueError as error:
        # @invariant: inv.successor_acyclic  (§5.3.1 R — AC §8.42 / SRC_SUCCESSOR_CYCLE)
        raise SourceValidationError(
            str(error), code=SRC_SUCCESSOR_CYCLE, records=[]
        ) from error
 
 
def _check_done_parent_terminal_children(categorized: dict[str, list]) -> None:
    status_by_id: dict[str, Any] = {}
    parent_of: dict[str, str] = {}
    for phase, _body in categorized.get("phases", []):
        status_by_id[phase.id] = phase.status
        for sub in phase.sub_phases:
            status_by_id[sub.id] = sub.status
            parent_of[sub.id] = phase.id
    for item, _body in categorized.get("items", []):
        status_by_id[item.id] = item.status
    for scanned, item, _body in categorized.get("promoted", []):
        status_by_id[item.id] = item.status
        parent_of[item.id] = scanned.phase_code
    for child_id, parent_id in parent_of.items():
        if status_by_id.get(parent_id) != "done":
            continue
        if status_by_id.get(child_id) not in TERMINAL_STATUSES:
            # @invariant: inv.done_parent_terminal_children  (§5.3.1 R — AC §8.50 / SRC_DONE_PARENT_LIVE_CHILD)
            raise SourceValidationError(
                f"done parent {parent_id!r} has non-terminal child {child_id!r} "
                f"(status={status_by_id.get(child_id)!r})",
                code=SRC_DONE_PARENT_LIVE_CHILD,
                records=[parent_id, child_id],
            )
 
 
def _check_order_and_code_uniqueness(categorized: dict[str, list]) -> None:
    for kind in ("eras", "tracks"):
        seen_order: dict[int, str] = {}
        for model in categorized.get(kind, []):
            code = getattr(model, "id", None) or getattr(model, "track_code", None)
            if model.order in seen_order:
                # @invariant: inv.era_track_order_unique  (§5.3.1 R — AC §8.56 / SRC_DUPLICATE_ORDER)
                raise SourceValidationError(
                    f"{kind}: duplicate order {model.order} ({seen_order[model.order]}, {code})",
                    code=SRC_DUPLICATE_ORDER,
                    records=[seen_order[model.order], str(code)],
                )
            seen_order[model.order] = str(code)
    seen_track: set[str] = set()
    for track in categorized.get("tracks", []):
        if track.track_code in seen_track:
            # @invariant: inv.phase_track_code_unique  (§5.3.1 R — AC §8.56 / SRC_DUPLICATE_CODE)
            raise SourceValidationError(
                f"duplicate track_code {track.track_code!r}",
                code=SRC_DUPLICATE_CODE,
                records=[track.track_code],
            )
        seen_track.add(track.track_code)
    seen_phase: set[str] = set()
    for phase, _body in categorized.get("phases", []):
        if phase.id in seen_phase:
            # @invariant: inv.phase_track_code_unique  (§5.3.1 R — AC §8.56 / SRC_DUPLICATE_CODE)
            raise SourceValidationError(
                f"duplicate phase_code {phase.id!r}",
                code=SRC_DUPLICATE_CODE,
                records=[phase.id],
            )
        seen_phase.add(phase.id)
 
 
def _check_version_metadata_agreement(categorized: dict[str, list]) -> None:
    by_version: dict[str, list] = {}
    for version in categorized.get("versions", []):
        by_version.setdefault(version.version, []).append(version)
    for version_id, records in by_version.items():
        if len(records) < 2:
            continue
        for field in _VERSION_AGREE_FIELDS:
            present = {getattr(rec, field) for rec in records if getattr(rec, field) is not None}
            if len(present) > 1:
                # @invariant: inv.version_metadata_agreement  (§5.3.1 R — AC §8.38 / SRC_VERSION_METADATA_CONFLICT)
                raise SourceValidationError(
                    f"version {version_id}: conflicting {field} across kinds: {sorted(map(str, present))}",
                    code=SRC_VERSION_METADATA_CONFLICT,
                    records=[version_id, field],
                )
 
 
def _check_roadmap_changelog_seq(categorized: dict[str, list]) -> None:
    """Every ``kind=roadmap`` changelog record must carry a unique positive-int
    ``extensions.changelog_seq`` (Plan 142 P6 / F1).
 
    The roadmap changelog is rendered by tiling the verbatim record bodies in seq
    order (byte-faithful to the authoritative history). A missing / non-int /
    non-positive / duplicate seq is **fail-closed** here so the renderer can never
    silently drop a record. ``kind=product`` records are unaffected (they render
    structured, ordered by version)."""
    seen: dict[int, str] = {}
    for version in categorized.get("versions", []):
        if getattr(version, "kind", None) != "roadmap":
            continue
        location = f"changelog/roadmap/{getattr(version, 'version', '?')}"
        seq = (getattr(version, "extensions", None) or {}).get("changelog_seq")
        if isinstance(seq, bool) or not isinstance(seq, int) or seq <= 0:
            # P6/F1 render-protection (NOT a §5.3.1 DDL-preservation @invariant —
            # changelog_seq lives in the open extensions dict; SRC_ROADMAP_CHANGELOG_SEQ).
            raise SourceValidationError(
                f"{location}: kind=roadmap record needs a positive-int "
                f"extensions.changelog_seq (got {seq!r})",
                code=SRC_ROADMAP_CHANGELOG_SEQ,
                records=[location],
                hint="every roadmap changelog record is a verbatim sequenced segment; "
                "append with max(changelog_seq)+10 per kind-dir (highest = newest, "
                "render descending); a mid-history entry uses the spec §3.5 "
                "mid-insert/renumber",
            )
        if seq in seen:
            # P6/F1 render-protection (NOT a §5.3.1 DDL-preservation @invariant —
            # changelog_seq lives in the open extensions dict; SRC_ROADMAP_CHANGELOG_SEQ).
            raise SourceValidationError(
                f"duplicate changelog_seq {seq} on {seen[seq]!r} and {location!r}",
                code=SRC_ROADMAP_CHANGELOG_SEQ,
                records=[seen[seq], location],
            )
        seen[seq] = location
 
 
def _check_product_changelog_seq(categorized: dict[str, list]) -> None:
    """The ``kind=product`` changelog seq is ALL-OR-NONE, via the shared validator
    (Plan 142 P2 / SRC_PRODUCT_CHANGELOG_SEQ).
 
    No product record carries a seq → the structured version-ordered fallback (OK).
    Any product record carries a seq → **every** selected product record must carry a
    unique positive-int ``extensions.changelog_seq`` (``bool`` rejected before
    ``int``); a missing / non-int / non-positive / duplicate seq in a sequenced set is
    **fail-closed** here so the verbatim render can never silently drop a record. The
    render path (``render_root_changelog``) calls the SAME ``check_changelog_seq``
    validator, so the regenerate gate and the renderer never drift."""
    located = [
        (
            f"changelog/product/{getattr(version, 'version', '?')}",
            (getattr(version, "extensions", None) or {}).get("changelog_seq"),
        )
        for version in categorized.get("versions", [])
        if getattr(version, "kind", None) == "product"
    ]
    problem = check_changelog_seq(located)
    if problem:
        # P2 render-protection (NOT a §5.3.1 DDL-preservation @invariant —
        # changelog_seq lives in the open extensions dict; SRC_PRODUCT_CHANGELOG_SEQ).
        raise SourceValidationError(
            problem,
            code=SRC_PRODUCT_CHANGELOG_SEQ,
            records=[loc for loc, _seq in located],
            hint="kind=product changelog seq is all-or-none; give every sequenced "
            "product record a unique positive-int extensions.changelog_seq",
        )
 
 
def _check_framing_anchors(categorized: dict[str, list]) -> None:
    """[NEW·V24] Validate every ``framing`` entity's placement anchor (§3.3).
 
    Each anchor must resolve: ``era`` → a known era id; ``version`` → a known
    changelog ``slug``; ``position`` → the ``top``/``bottom`` sentinel or a sibling
    record id (a version slug or another framing id). An unresolved target is
    ``SRC_FRAMING_DANGLING_ANCHOR``. Two framings sharing the same
    ``(target_kind, target_id, placement, order)`` — **including both order-unset**
    (``None`` is a value here) — is ``SRC_FRAMING_DUPLICATE_ANCHOR`` (§3.3, no
    further id tiebreak). These are source-only (framing has no store rows), so
    the check runs over the loaded models."""
    framings = categorized.get("framing", [])
    if not framings:
        return
    era_ids = {era.id for era in categorized.get("eras", [])}
    version_slugs = {version.slug for version in categorized.get("versions", [])}
    framing_ids = {framing.id for framing in framings}
    position_ids = {"top", "bottom"} | version_slugs | framing_ids
    resolvable = {"era": era_ids, "version": version_slugs, "position": position_ids}
    seen: dict[tuple, str] = {}
    for framing in framings:
        anchor = framing.anchor
        location = f"framing/{framing.id}.md"
        if anchor.target_id not in resolvable[anchor.target_kind]:
            raise SourceValidationError(
                f"{location}: framing anchor references unknown "
                f"{anchor.target_kind} target {anchor.target_id!r}",
                code=SRC_FRAMING_DANGLING_ANCHOR,
                records=[location],
                hint="anchor.target_id must resolve to an era id / version slug / "
                "position sibling id or the top|bottom sentinel",
            )
        key = (anchor.target_kind, anchor.target_id, anchor.placement, anchor.order)
        if key in seen:
            raise SourceValidationError(
                f"duplicate framing anchor {key} on {seen[key]!r} and {location!r}",
                code=SRC_FRAMING_DUPLICATE_ANCHOR,
                records=[seen[key], location],
                hint="two framings share the same (target, placement, order) — "
                "give one an explicit distinct order",
            )
        seen[key] = location
 
 
def validate_cross_record(categorized: dict[str, list]) -> None:
    """Run every §5.3 step-6 cross-record check (raises on the first failure)."""
    func = "validate_cross_record"
    items = _item_models(categorized)
    id_index = _check_id_uniqueness(items)
    _check_depends_on_and_cycle(items, id_index)
    _check_successor_acyclicity(items)
    _check_done_parent_terminal_children(categorized)
    _check_order_and_code_uniqueness(categorized)
    _check_version_metadata_agreement(categorized)
    _check_roadmap_changelog_seq(categorized)
    _check_product_changelog_seq(categorized)
    _check_framing_anchors(categorized)
    _dbg("OC_CHAT.MAIN", func, f"cross-record ok for {len(items)} items", 3)