verify_142_seed_fidelity.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tools/verify_142_seed_fidelity.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/seed/test_seed_142.py"
"""
The one-time seed-fidelity verifier (Plan 142 P1.7, §9 — CI-runnable).
 
Purpose
-------
Verify the committed ``oc_chat/roadmap/source/**`` seed against the live v3.38
oracle (Registry Source & Storage Spec V0.22 §9): the source is canonical
(``format --check``); ``regenerate`` is deterministic + idempotent (the
byte-stable source->store property — §9.2); the migration is **lossless** (every
v3.38 item_id appears in the regenerated store, modulo the classified dedup —
zero unclassified drops, §9.4); and it quantifies the §5.6 sub-phase promotions
(§9 3c). The render-based 4-tuple (§9.3) + the §9 (1) renderer-diff + (3b)
ordering are operator-staged precursors (flagged, not self-attested).
 
Run from the worktree root, preferably as ``python -m tools.verify_142_seed_fidelity``
[roadmap_dir] [source_dir]; the direct ``python tools/verify_142_seed_fidelity.py``
also works via the ``sys.path`` bootstrap below.
 
Responsibilities
----------------
- ``verify(roadmap_dir, source_dir)`` -> a structured result (CI exit 0/1).
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - verification lifecycle
    L3 - per-check results
    L4 - per-record comparison
 
Contracts
---------
- Read-only over the committed seed; the oracle + regenerated stores are
  worktree-local temporaries; never touches the shared registry.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- tools.seed_142_ingest (ingest_oracle and emit_source - builds the v3.38 oracle store and re-emits it for the promotion and dropped-dependency counts)
- oc_chat.registry.regenerate (imported for the contracts bootstrap it runs)
- oc_chat.registry.regenerate.pipeline (regenerate - rebuilds the store twice for the idempotence check)
- oc_chat.registry.source.canonical (format_source_tree - the canonical-source check in report mode)
- oc_chat.registry.source.scanner (scan_source_tree - lists the committed source records)
"""
from __future__ import annotations
 
import json
import re
import sqlite3
import sys
import tempfile
from dataclasses import dataclass, field
from pathlib import Path
 
# Force THIS worktree's ``oc_chat`` + ``tools`` ahead of the editable main-repo
# install, so a direct ``python tools/verify_142_seed_fidelity.py`` resolves the
# worktree (otherwise ``from tools.seed_142_ingest import …`` -> ModuleNotFoundError).
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
from oc_diagnostics import _dbg
 
import oc_chat.registry.regenerate  # noqa: F401  (runs the contracts bootstrap)
 
from oc_chat.registry.regenerate.pipeline import regenerate
from oc_chat.registry.source.canonical import format_source_tree
from oc_chat.registry.source.scanner import scan_source_tree
from tools.seed_142_ingest import emit_source, ingest_oracle
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
_DEFAULT_ROADMAP = Path("oc_chat/roadmap")
_DEFAULT_SOURCE = Path("oc_chat/roadmap/source")
 
 
@dataclass
class FidelityResult:
    ok: bool = True
    checks: dict = field(default_factory=dict)
    promotions: int = 0
    classified_drops: int = 0
    unclassified_drops: list = field(default_factory=list)
    notes: list = field(default_factory=list)
 
    def fail(self, name: str, detail: object) -> None:
        self.ok = False
        self.checks[name] = {"pass": False, "detail": detail}
 
    def passed(self, name: str, detail: object = None) -> None:
        self.checks[name] = {"pass": True, "detail": detail}
 
 
def _sanitize_id(raw: str) -> str:
    candidate = raw.replace("–", "-").replace("—", "-")
    candidate = re.sub(r"[^A-Za-z0-9._-]", "_", candidate).strip("._-")
    return candidate or "x"
 
 
def _store_item_ids(db_path: Path) -> set[str]:
    conn = sqlite3.connect(str(db_path))
    try:
        return {row[0] for row in conn.execute("SELECT item_id FROM roadmap_items_v3")}
    finally:
        conn.close()
 
 
def verify(roadmap_dir: Path = _DEFAULT_ROADMAP, source_dir: Path = _DEFAULT_SOURCE) -> FidelityResult:
    """Run the §9 CI fidelity checks against the committed seed."""
    func = "verify"
    result = FidelityResult()
    source_dir = Path(source_dir)
    with tempfile.TemporaryDirectory() as raw_tmp:
        tmp = Path(raw_tmp)
        # Oracle: ingest the live v3.38 set + re-emit (for the migration report / promotions).
        rdir = tmp / "roadmap"
        rdir.mkdir()
        for name in ("ROADMAP-v3.38.md", "ROADMAP-EXEC-ORDER-v3.38.md",
                     "ROADMAP-CHANGELOG-v3.38.md", "ROADMAP-TODO-v3.38.md"):
            (rdir / name).write_text((Path(roadmap_dir) / name).read_text(encoding="utf-8"), encoding="utf-8")
        oracle_db = tmp / "oracle.sqlite"
        ingest_oracle(rdir, oracle_db)
        oracle_ids = {_sanitize_id(i) for i in _store_item_ids(oracle_db)}
        regen_report = emit_source(oracle_db, tmp / "reemit_source", Path(roadmap_dir) / "ROADMAP-v3.38.md")
        result.promotions = len(regen_report.promotions)
        result.classified_drops = len(regen_report.dropped_parent_child_deps)
 
        # Check 1 — the committed source is canonical (format --check).
        records = scan_source_tree(source_dir)
        non_canonical = format_source_tree(records, check=False)  # report mode (no rewrite)
        if non_canonical:
            result.fail("source_canonical", non_canonical[:10])
        else:
            result.passed("source_canonical", f"{len(records)} records canonical")
 
        # Check 2 — regenerate is deterministic + idempotent (byte-stable source->store).
        seed_db = tmp / "seed.sqlite"
        first = regenerate(source_dir, seed_db)
        if not first.ok:
            result.fail("regenerate", first.envelope)
            _dbg("OC_CHAT.MAIN", func, "regenerate failed", 2)
            return result
        second = regenerate(source_dir, seed_db)
        if first.source_hash == second.source_hash and first.logical_state_hash == second.logical_state_hash:
            result.passed("regenerate_idempotent", {"source_hash": first.source_hash,
                                                     "logical_state_hash": first.logical_state_hash})
        else:
            result.fail("regenerate_idempotent", "hashes differ across rebuilds")
 
        # Check 3 — lossless: every v3.38 item appears in the regenerated store (zero unclassified drops).
        regen_ids = _store_item_ids(seed_db)
        missing = sorted(oracle_ids - regen_ids)
        if missing:
            result.unclassified_drops = missing
            result.fail("lossless_items", {"missing": missing[:20], "count": len(missing)})
        else:
            result.passed("lossless_items", f"{len(oracle_ids)} v3.38 items all present")
 
    # §9 operator-staged precursors (NOT self-attested here).
    result.notes = [
        "§9 (1) renderer-diff vs the current files is operator-staged (run after P1.6 render lands on the seed).",
        "§9 (3) render-based 4-tuple is operator-staged; this verifier proves store-level losslessness instead.",
        "§9 (3b) per-phase item-ordering-matches-intent is operator-staged.",
        f"§9 (3c) sub-phase promotions: {result.promotions}; classified parent-child dep drops: {result.classified_drops}.",
    ]
    _dbg("OC_CHAT.MAIN", func, f"fidelity ok={result.ok}", 2)
    return result
 
 
def main(argv: list[str] | None = None) -> int:
    """CI entry point: exit 0 if all gates pass, 1 otherwise."""
    args = argv if argv is not None else sys.argv[1:]
    roadmap = Path(args[0]) if len(args) > 0 else _DEFAULT_ROADMAP
    source = Path(args[1]) if len(args) > 1 else _DEFAULT_SOURCE
    result = verify(roadmap, source)
    print(json.dumps({
        "ok": result.ok, "checks": result.checks, "promotions": result.promotions,
        "classified_drops": result.classified_drops, "unclassified_drops": result.unclassified_drops,
        "notes": result.notes,
    }, indent=2, sort_keys=True))
    return 0 if result.ok else 1
 
 
if __name__ == "__main__":
    raise SystemExit(main())