seed_142_product_changelog.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tools/seed_142_product_changelog.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/seed/test_product_changelog.py"
"""
Seed the live root product CHANGELOG.md into kind=product source records (Plan 142).
 
Purpose
-------
The live repo-root product ``CHANGELOG.md`` is not registry-managed (0 ``kind=product``
records → a 296-byte stub render). Tile it **content-faithfully** into
``source/changelog/product/<id>.md`` records: one ``kind=product`` record per
separator-delimited segment, body = the verbatim slice (canonicalised at
``format_record_text`` time), so the verbatim-seq render reproduces
``canonicalize(backup)`` byte-for-byte. The raw original survives as the Phase-0
backup receipt; this writes ONLY ``source/changelog/product/**`` (the roadmap seed
is never touched), and never the live ``CHANGELOG.md``.
 
Deliberately NOT the roadmap reseed: no destructive ``changelog/`` ``rglob`` cleanup,
no ``_BUMP_RE`` arrow classifier. Keys are **independent synthetic** versions in a
``v0.0.x`` band proven disjoint from every version in the source tree (all kinds) +
the store, fail-closed (mirroring a roadmap bump arrow would mint ``v3.03–v3.10`` and
hard-fail at ``SRC_VERSION_METADATA_CONFLICT`` or silently dedup-merge).
 
Responsibilities
----------------
- ``backfill`` (the migration), ``synthetic_versions`` (disjoint band, fail-closed),
  ``_roadmap_snapshot`` (the untouched guard), reported ``SeedReport`` for the audit.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - seed lifecycle
    L3 - per-segment record emission
    L4 - version/date assignment
 
Contracts
---------
- Writes ONLY ``source/changelog/product/**``; a pre/post hash+filelist guard over
  ``source/changelog/roadmap/**`` fails closed on any change. Idempotent: a non-empty
  product dir must match this run's expected record set or it fails closed.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- option_c_registry_contracts (the Version model and SCHEMA_VERSION for each product record)
- oc_chat.registry.regenerate (imported for the contracts bootstrap it runs)
- oc_chat.registry.regenerate.pipeline (regenerate - rebuilds a temporary store for the fidelity proof)
- oc_chat.registry.render.roadmap (render_living_set - renders the root changelog for the fidelity proof)
- oc_chat.registry.source.canonical (format_record_text and load_record - writes each product record and reads existing versions)
- oc_chat.registry.source.canonical_product (split_segments, segment_line_ranges and product_changelog_canonicalize - tiles and canonicalises the changelog)
- oc_chat.registry.source.scanner (scan_source_tree - finds the version records whose versions the synthetic band avoids)
"""
from __future__ import annotations
 
import hashlib
import re
import sys
import tempfile
from dataclasses import dataclass, field
from pathlib import Path
 
# Force the WORKTREE-local oc_chat ahead of the editable main-repo install.
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
from oc_diagnostics import _dbg
 
import oc_chat.registry.regenerate  # noqa: F401  (contracts bootstrap)
import option_c_registry_contracts as rc
 
from oc_chat.registry.regenerate.pipeline import regenerate
from oc_chat.registry.render.roadmap import render_living_set
from oc_chat.registry.source.canonical import format_record_text, load_record
from oc_chat.registry.source.canonical_product import (
    product_changelog_canonicalize,
    segment_line_ranges,
    split_segments,
)
from oc_chat.registry.source.scanner import scan_source_tree
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
#: Synthetic product-record metadata profile (mirrors the roadmap reseed's date/era).
_ERA = "v3"
#: NAMED date fallback for a segment carrying no inline ISO date (recorded in coverage).
_FALLBACK_DATE = "2026-05-15"
#: The reserved synthetic-version band floor (the roadmap synthetic band runs dense
#: to v0.0.58; 1000+ is well clear — but disjointness is asserted at runtime).
_SYNTH_FLOOR = 1000
_ISO_DATE_RE = re.compile(r"\b(\d{4}-\d{2}-\d{2})\b")
 
 
@dataclass
class SeedReport:
    records_written: int = 0
    segments: int = 0
    synthetic_band: tuple[str, str] = ("", "")
    date_fallback_seqs: list[int] = field(default_factory=list)
    canonical_bytes: int = 0
    rows: list[dict] = field(default_factory=list)  # per-segment audit
 
 
def _roadmap_snapshot(roadmap_dir: Path) -> dict[str, str]:
    """``{relpath: sha256}`` over ``source/changelog/roadmap/**`` (the untouched guard)."""
    out: dict[str, str] = {}
    if roadmap_dir.is_dir():
        for path in sorted(roadmap_dir.rglob("*.md")):
            out[path.name] = hashlib.sha256(path.read_bytes()).hexdigest()
    return out
 
 
def _gather_existing_versions(source_dir: Path) -> set[str]:
    """Every NON-product ``version`` string in the scanned source tree.
 
    The synthetic band must be disjoint from the roadmap (and any other non-product)
    version set — colliding there hard-fails at ``SRC_VERSION_METADATA_CONFLICT`` or
    silently dedup-merges. The product records this seed (re)writes are deliberately
    EXCLUDED so the band is stable across re-runs (idempotent); product versions are
    unique within the band by construction.
    """
    versions: set[str] = set()
    for scanned in scan_source_tree(source_dir):
        if scanned.kind == "version":
            model, _body = load_record(scanned)
            if model.kind != "product":
                versions.add(model.version)
    return versions
 
 
def synthetic_versions(count: int, existing: set[str]) -> list[str]:
    """A deterministic ``v0.0.<n>`` band of ``count`` versions, provably disjoint
    from ``existing`` (fail-closed if no disjoint band is found at the floor)."""
    base = _SYNTH_FLOOR
    while any(f"v0.0.{base + i}" in existing for i in range(count)):
        base += count
    band = [f"v0.0.{base + i}" for i in range(count)]
    overlap = set(band) & existing
    if overlap:  # fail-closed (cannot happen given the loop, but assert the invariant)
        raise ValueError(f"synthetic version band collides with existing versions: {sorted(overlap)}")
    return band
 
 
def _first_iso_date(segment: str) -> str | None:
    match = _ISO_DATE_RE.search(segment)
    return match.group(1) if match else None
 
 
def backfill(backup_path: Path, source_dir: Path) -> SeedReport:
    """Tile ``backup_path`` into ``source/changelog/product/**`` records."""
    func = "backfill"
    source_dir = Path(source_dir)
    roadmap_dir = source_dir / "changelog" / "roadmap"
    product_dir = source_dir / "changelog" / "product"
 
    before = _roadmap_snapshot(roadmap_dir)
    raw = Path(backup_path).read_text(encoding="utf-8")
    segments = split_segments(raw)
    line_ranges = segment_line_ranges(raw)
    canonical = product_changelog_canonicalize(raw)
 
    existing = _gather_existing_versions(source_dir)
    band = synthetic_versions(len(segments), existing)
 
    # Plan the records before writing (for the idempotency check).
    planned: dict[str, str] = {}  # filename -> file text
    report = SeedReport(segments=len(segments), canonical_bytes=len(canonical.encode("utf-8")))
    report.synthetic_band = (band[0], band[-1])
    for index, (segment, version, (lo, hi)) in enumerate(zip(segments, band, line_ranges), start=1):
        date = _first_iso_date(segment)
        if date is None:
            date = _FALLBACK_DATE
            report.date_fallback_seqs.append(index)
        model = rc.Version(
            schema_version=rc.SCHEMA_VERSION,
            version=version,
            date=date,
            era=_ERA,
            kind="product",
            extensions={"changelog_seq": index},
        )
        text = format_record_text(model, body=segment)
        filename = f"{index:03d}-{version}.md"
        planned[filename] = text
        body_hash = hashlib.sha256(text.encode("utf-8")).hexdigest()
        report.rows.append({
            "seq": index, "version": version, "lines": f"{lo}-{hi}",
            "destination": f"changelog/product/{filename}", "date": date,
            "body_sha256": body_hash,
        })
 
    # Idempotency: a non-empty product dir must match exactly what we plan to write.
    if product_dir.is_dir():
        on_disk = {p.name: p.read_text(encoding="utf-8") for p in product_dir.glob("*.md")}
        if on_disk and on_disk != planned:
            raise ValueError(
                "product changelog dir is non-empty and does not match this seed's expected "
                f"record set (on-disk {len(on_disk)} vs planned {len(planned)}); refusing (fail-closed)"
            )
 
    product_dir.mkdir(parents=True, exist_ok=True)
    for filename, text in planned.items():
        (product_dir / filename).write_text(text, encoding="utf-8", newline="")
        report.records_written += 1
 
    after = _roadmap_snapshot(roadmap_dir)
    if before != after:
        raise ValueError("roadmap changelog records changed during product seed (fail-closed)")
 
    # Self-validating fidelity: the verbatim-seq render must reproduce
    # canonicalize(backup) byte-for-byte (header-stripped). Fail-closed.
    with tempfile.TemporaryDirectory() as tmp:
        check_db = Path(tmp) / "seed_fidelity.sqlite"
        result = regenerate(source_dir, check_db)
        if not result.ok:
            raise ValueError(f"regenerate failed after product seed (fail-closed): {result.envelope}")
        rendered = render_living_set(check_db, source_dir)["root_changelog"]
    body = rendered.split("-->", 1)[1].lstrip("\n")
    if body != canonical:
        raise ValueError("render(seed) != canonicalize(backup) — content-faithful proof failed (fail-closed)")
 
    _dbg("OC_CHAT.MAIN", func, f"wrote {report.records_written} product records; render==canon", 2)
    return report
 
 
def _main() -> int:
    func = "_main"
    repo = Path(__file__).resolve().parents[1]
    backup = repo / "CHANGELOG.md"  # the stable live oracle (byte-identical to the backup)
    source = repo / "oc_chat" / "roadmap" / "source"
    report = backfill(backup, source)
    print(f"records_written={report.records_written} segments={report.segments}")
    print(f"synthetic_band={report.synthetic_band[0]}..{report.synthetic_band[1]}")
    print(f"date_fallback_seqs={report.date_fallback_seqs}")
    print(f"canonical_bytes={report.canonical_bytes}")
    _dbg("OC_CHAT.MAIN", func, f"records={report.records_written}", 2)
    return 0
 
 
if __name__ == "__main__":
    raise SystemExit(_main())