seed_142_prose_backfill.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tools/seed_142_prose_backfill.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/seed/test_prose_backfill.py"
"""
Per-phase prose backfill (Plan 142 prose-preservation Phase 2, GATE2 step 1).
 
Purpose
-------
P1.7 seeded the roadmap ITEM SET into the registry but dropped the per-phase
PROSE — every ``# PHASE Mx`` section in the legacy
``oc_chat/roadmap/ROADMAP-v3.38.md`` (phase metadata + the full ``### Deliverables``
table with each item's *rich description + status provenance* + ``### Acceptance``)
was lost: the seeded ``phases/<code>.md`` records carry an EMPTY body and a few
``sub_phases`` descriptions degraded to bare-id stubs (e.g. ``M0.2``).
 
This tool restores them, byte-faithfully, mirroring the Phase-1 pattern (verbatim
section -> record body): for each ``# PHASE`` section it writes the verbatim
section body into the matching ``phases/<code>.md`` record BODY (which the P1.4
build already carries into ``roadmap_items_v3.prose`` via ``prose=body or None``,
so the EXTENDED render can emit it). It also replaces each ``description == id``
sub-phase stub with the real Item-column text from that phase's deliverables
table (single-lined, capped to the ``Line240`` contract).
 
No contract change: the Phase model is unchanged; only record BODIES (and the
handful of stub descriptions) are rewritten. Idempotent — re-running yields
identical bytes.
 
Responsibilities
----------------
- ``parse_phase_sections`` (split the legacy file on ``# PHASE`` headers),
  ``deliverable_items`` (id -> Item-column text), ``backfill`` (re-emit records),
  ``coverage`` (the per-phase / per-item map for the audit artifact).
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - backfill lifecycle
    L3 - per-phase section extraction
    L4 - per-record emission
 
Contracts
---------
- Deterministic; writes only ``source/phases/<code>.md`` records; never touches
  the live legacy roadmap; every rewritten record re-validates + re-canonicalises.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.regenerate (imported for the contracts bootstrap it runs)
- oc_chat.registry.source.canonical (load_record and format_record_text - reads each phase record and re-emits it with the restored body)
- oc_chat.registry.source.scanner (scan_source_tree - finds the phase records)
"""
from __future__ import annotations
 
import re
import sys
from dataclasses import dataclass, field
from pathlib import Path
 
# Force the WORKTREE-local oc_chat ahead of the editable main-repo install (a
# tools/ script's sys.path[0] is tools/, not the worktree root).
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
from oc_diagnostics import _dbg
 
import oc_chat.registry.regenerate  # noqa: F401  (contracts bootstrap)
 
from oc_chat.registry.source.canonical import format_record_text, load_record
from oc_chat.registry.source.scanner import scan_source_tree
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
#: A ``# PHASE <code> — <title>`` header line (em-dash separated, col 0).
_PHASE_HEADER_RE = re.compile(r"^# PHASE (\S+) — (.+)$", re.MULTILINE)
#: The next top-level boundary after a phase section (next phase / next ``## ``).
_BOUNDARY_RE = re.compile(r"^(# PHASE |## )", re.MULTILINE)
#: A trailing horizontal-rule separator emitted between phase sections.
_TRAILING_RULE_RE = re.compile(r"\n*-{3,}\s*$")
#: A deliverables-table data row (a leading-id cell, not the header/separator).
_DELIV_ROW_RE = re.compile(r"^\|\s*(?P<id>[A-Za-z][\w.\-]*)\s*\|\s*(?P<item>.+?)\s*\|.*\|\s*$")
 
_LINE240_MAX = 240
 
 
@dataclass
class BackfillReport:
    phases_written: int = 0
    bodies: dict = field(default_factory=dict)        # code -> verbatim body
    stubs_fixed: list = field(default_factory=list)   # (code, sub_id, new_desc)
    skipped: list = field(default_factory=list)       # codes with no legacy section
 
 
def parse_phase_sections(roadmap_text: str) -> dict[str, str]:
    """Return ``{phase_code: verbatim_section_body}`` for every ``# PHASE`` section.
 
    The body is everything after the header line up to the next ``# PHASE`` /
    top-level ``## `` heading, with the inter-section ``---`` rule stripped.
    """
    text = roadmap_text.replace("\r\n", "\n")
    out: dict[str, str] = {}
    for match in _PHASE_HEADER_RE.finditer(text):
        code = match.group(1)
        nxt = _BOUNDARY_RE.search(text, match.end())
        end = nxt.start() if nxt else len(text)
        body = text[match.end():end].strip("\n")
        body = _TRAILING_RULE_RE.sub("", body).strip("\n")
        out[code] = body
    return out
 
 
def deliverable_items(section_body: str) -> dict[str, str]:
    """Map ``sub_id -> Item-column text`` from a phase section's deliverables table."""
    items: dict[str, str] = {}
    for line in section_body.split("\n"):
        stripped = line.strip()
        if not stripped.startswith("|") or set(stripped) <= set("|-: "):
            continue  # not a row, or the header separator
        match = _DELIV_ROW_RE.match(stripped)
        if match is None:
            continue
        item_id = match.group("id").strip()
        item_text = " ".join(match.group("item").split()).strip()
        if item_id.lower() in ("#", "item") or not item_text:
            continue  # the table header row
        items[item_id] = item_text
    return items
 
 
def _capped(text: str) -> str:
    """Single-line + cap to the ``Line240`` contract (whole-word trim)."""
    text = " ".join(text.split())
    if len(text) <= _LINE240_MAX:
        return text
    clipped = text[: _LINE240_MAX - 1].rstrip()
    return (clipped[: clipped.rfind(" ")] if " " in clipped else clipped).rstrip() + "…"
 
 
def backfill(roadmap_md: Path, source_dir: Path) -> BackfillReport:
    """Backfill every ``phases/<code>.md`` body from the legacy ``# PHASE`` sections."""
    func = "backfill"
    sections = parse_phase_sections(Path(roadmap_md).read_text(encoding="utf-8"))
    report = BackfillReport()
    for scanned in scan_source_tree(Path(source_dir)):
        if scanned.kind != "phase":
            continue
        code = scanned.stem
        body = sections.get(code)
        if body is None:
            report.skipped.append(code)
            continue
        model, _old_body = load_record(scanned)
        items = deliverable_items(body)
        new_subs = []
        for sub in model.sub_phases:
            if sub.description == sub.id and sub.id in items:  # frozen models: copy-update
                new_desc = _capped(items[sub.id])
                new_subs.append(sub.model_copy(update={"description": new_desc}))
                report.stubs_fixed.append((code, sub.id, new_desc))
            else:
                new_subs.append(sub)
        model = model.model_copy(update={"sub_phases": new_subs})
        scanned.path.write_text(
            format_record_text(model, body=body), encoding="utf-8", newline=""
        )
        report.phases_written += 1
        report.bodies[code] = body
        _dbg("OC_CHAT.MAIN", func, f"backfilled {code} ({len(body)} chars)", 3)
    _dbg("OC_CHAT.MAIN", func, f"wrote {report.phases_written} phase bodies", 2)
    return report
 
 
def _main() -> int:
    func = "_main"
    repo = Path(__file__).resolve().parents[1]
    roadmap = repo / "oc_chat" / "roadmap" / "ROADMAP-v3.38.md"
    source = repo / "oc_chat" / "roadmap" / "source"
    report = backfill(roadmap, source)
    _dbg("OC_CHAT.MAIN", func, f"phases={report.phases_written} stubs={len(report.stubs_fixed)} "
         f"skipped={report.skipped}", 2)
    print(f"phases_written={report.phases_written}")
    print(f"stubs_fixed={report.stubs_fixed}")
    print(f"skipped (no legacy # PHASE section)={report.skipped}")
    return 0
 
 
if __name__ == "__main__":
    raise SystemExit(_main())