seed_142_ingest.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tools/seed_142_ingest.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/seed/test_seed_142.py"
"""
The v3.38 binary -> AuthoredSource seed migration (Plan 142 P1.7, §9 / §5.8).
 
Purpose
-------
Ingest the live ``ROADMAP-v3.38*`` four-file set into a worktree-local store
(the oracle), then EMIT the per-record ``oc_chat/roadmap/source/**`` tree: 14
top-level phases as ``Phase`` records (sub-phases nested as ``sub_phases`` or
promoted to ``phases/<code>/<sub_id>.md`` per §5.6), non-phase items as
``RoadmapItem`` records, plus the era/tracks/version — backfilling a non-empty
``purpose`` for every phase + item (AC §8.22) and sanitising legacy ids/dates.
 
Responsibilities
----------------
- ``ingest_oracle(roadmap_dir, db)`` — build the oracle store.
- ``emit_source(db, out_dir, roadmap_md)`` — write the canonical source tree;
  return a migration report (counts, promotions, id/date sanitisations).
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - migration lifecycle
    L3 - per-entity emission
    L4 - per-record sanitisation
 
Contracts
---------
- Deterministic: identical inputs -> identical source bytes.
- Never writes outside ``out_dir``; the oracle DB is worktree-local + temporary.
- Every emitted record validates against the P1.1 / P1.4 models.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.regenerate (imported for the contracts bootstrap it runs)
- option_c_registry_contracts (the record models, SCHEMA_VERSION, ID_RE and TERMINAL_STATUSES for the emitted source tree)
- oc_chat.registry (Registry - the oracle registry over the SQLite backend)
- oc_chat.registry.backends.sqlite (SQLiteBackend and RegistrySeed - the worktree-local oracle store)
- oc_chat.registry.regenerate.promoted (PromotedItem - the record for a sub-phase promoted to its own file)
- oc_chat.registry.roadmap.ingest (ingest_corpus - ingests the v3.38 four-file set into the oracle)
- oc_chat.registry.source.canonical (format_record_text - writes each emitted record canonically)
"""
from __future__ import annotations
 
import hashlib
import re
import sqlite3
import uuid
from dataclasses import dataclass, field
from pathlib import Path
 
from oc_diagnostics import _dbg
 
import oc_chat.registry.regenerate  # noqa: F401  (runs the contracts bootstrap)
import option_c_registry_contracts as rc
 
from oc_chat.registry import Registry
from oc_chat.registry.backends.sqlite import RegistrySeed, SQLiteBackend
from oc_chat.registry.regenerate.promoted import PromotedItem
from oc_chat.registry.roadmap.ingest import ingest_corpus
from oc_chat.registry.source.canonical import format_record_text
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
_ISO_DATE_RE = re.compile(r"(\d{4}-\d{2}-\d{2})")
_PHASE_HEADER_RE = re.compile(
    # OCT-LINT: disable=no_hardcoded_secrets reason="Seed migration constants + placeholder slot are not secrets"
    r"^#\s+PHASE\s+(?P<code>M\S+)\s*[—-]\s*(?P<title>.+?)\s*$", re.MULTILINE
)
_ID_OK_RE = re.compile(rc.ID_RE)
_PRIORITIES = {"critical", "high", "medium", "lower"}
_DEFAULT_DATE = "2026-06-02"
 
 
@dataclass
class MigrationReport:
    counts: dict = field(default_factory=dict)
    promotions: list = field(default_factory=list)
    id_sanitisations: list = field(default_factory=list)
    purpose_backfilled: int = 0
    dropped_parent_child_deps: list = field(default_factory=list)
    reconciled_statuses: list = field(default_factory=list)
 
 
def _one_line(text: object, max_len: int = 240) -> str:
    collapsed = " ".join(str(text or "").split())
    return collapsed[:max_len].strip()
 
 
def _iso_date(value: object) -> str:
    match = _ISO_DATE_RE.search(str(value or ""))
    return match.group(1) if match else _DEFAULT_DATE
 
 
def _sanitize_id(raw: str, report: MigrationReport) -> str:
    """Map a legacy id to the ASCII id shape (en/em-dash -> '-', illegal -> '_')."""
    candidate = raw.replace("–", "-").replace("—", "-")
    candidate = re.sub(r"[^A-Za-z0-9._-]", "_", candidate).strip("._-")
    if not candidate:
        candidate = "x"
    if candidate != raw:
        report.id_sanitisations.append({"from": raw, "to": candidate})
    return candidate
 
 
def _derive_purpose(title: object, prose: object, report: MigrationReport) -> str:
    """Non-empty single-line ≤240 purpose: prose first sentence -> title (AC §8.22)."""
    text = ""
    if prose and str(prose).strip():
        first = str(prose).strip().splitlines()[0]
        text = re.split(r"(?<=[.!?])\s", first)[0]
    if not text.strip():
        text = str(title or "")
    purpose = _one_line(text, 240)
    if not purpose:
        purpose = "Seed item (purpose pending operator backfill)"
    report.purpose_backfilled += 1
    return purpose
 
 
def phase_title_map(roadmap_md: Path) -> dict[str, str]:
    text = roadmap_md.read_text(encoding="utf-8")
    return {
        match.group("code"): _one_line(match.group("title"))
        for match in _PHASE_HEADER_RE.finditer(text)
    }
 
 
def ingest_oracle(roadmap_dir: Path, db_path: Path) -> Registry:
    backend = SQLiteBackend(db_path)
    backend.initialize(
        RegistrySeed(
            workspace_id=str(uuid.uuid4()), registry_id=str(uuid.uuid4()),
            workspace_root=str(db_path.parent), git_common_dir=str(db_path.parent / ".git"),
            registry_path=str(db_path), checkpoint_path=str(db_path.parent / ".option_c" / "cp.json"),
            operator_slot="operator-seed", hmac_key_id=str(uuid.uuid4()),
            secret_fingerprint=hashlib.sha256(b"seed").hexdigest(), app_version="seed",
        )
    )
    registry = Registry(backend, clock=backend.clock)
    result = ingest_corpus(registry, roadmap_dir, actor_slot="operator-seed")
    if not result.ok:
        raise RuntimeError(f"v3.38 ingest failed: {result.errors}")
    return registry
 
 
def _deps_and_tracks(conn: sqlite3.Connection, report: MigrationReport) -> tuple[dict, dict]:
    parent_of = {
        row["item_id"]: row["parent_item_id"]
        for row in conn.execute("SELECT item_id, parent_item_id FROM roadmap_items_v3")
    }
    deps: dict[str, list] = {}
    for row in conn.execute("SELECT from_item_id, to_item_id, kind FROM roadmap_item_dependencies"):
        pred, dep = row["from_item_id"], row["to_item_id"]
        # Drop parent-child depends_on edges: part_of is carried by parent_item_id
        # nesting (§3.2), never depends_on — and these form the v3.38 M11<->M11.8 cycle.
        if parent_of.get(dep) == pred or parent_of.get(pred) == dep:
            report.dropped_parent_child_deps.append({"from": pred, "to": dep, "kind": row["kind"]})
            continue
        deps.setdefault(dep, []).append((pred, row["kind"]))
    tracks: dict[str, list] = {}
    sql = ("SELECT it.item_id, t.track_code FROM roadmap_item_tracks it "
           "JOIN roadmap_tracks t ON t.track_id = it.track_id ORDER BY t.track_code")
    for row in conn.execute(sql):
        tracks.setdefault(row["item_id"], []).append(row["track_code"])
    return deps, tracks
 
 
def _depends_on(item_id: str, deps: dict, sanitized: dict) -> list:
    out: list = []
    for predecessor, kind in deps.get(item_id, []):
        pid = sanitized.get(predecessor, predecessor)
        out.append(pid if kind == "blocks" else rc.DependencyEdge(id=pid, kind=kind))
    return out
 
 
def _needs_promotion(row: sqlite3.Row, deps: dict) -> bool:
    """§9 3c: promote only on item-specific data the nested form cannot hold —
    description >240, a (non-inherited) priority, or dependencies. The inherited
    parent track + the global ``v3.38`` version stamp are NOT triggers (they are
    classified dedup drops, not per-item data)."""
    return bool(len((row["title"] or "")) > 240 or row["priority"] is not None or deps.get(row["item_id"]))
 
 
def _write(out_dir: Path, rel: str, model: rc.RegistryModel, body: str = "") -> None:
    path = out_dir / rel
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(format_record_text(model, body=body), encoding="utf-8", newline="")
 
 
def emit_source(db_path: Path, out_dir: Path, roadmap_md: Path) -> MigrationReport:
    """Emit the canonical source tree from the oracle store; return the report."""
    func = "emit_source"
    report = MigrationReport()
    titles = phase_title_map(roadmap_md)
    conn = sqlite3.connect(str(db_path))
    conn.row_factory = sqlite3.Row
    deps, tracks = _deps_and_tracks(conn, report)
    # Sanitise every item id up front so cross-references resolve consistently.
    sanitized = {
        row["item_id"]: _sanitize_id(row["item_id"], report)
        for row in conn.execute("SELECT item_id FROM roadmap_items_v3")
    }
 
    version_row = conn.execute(
        # OCT-LINT: disable=no_hardcoded_secrets reason="Seed migration constants + placeholder slot are not secrets"
        "SELECT version_id FROM roadmap_versions ORDER BY status='current' DESC, version_id DESC LIMIT 1"
    ).fetchone()
    roadmap_version = version_row["version_id"] if version_row else "v3.38"
    _write(out_dir, "manifest.md", rc.Manifest(
        schema_version=rc.SCHEMA_VERSION, roadmap_version=roadmap_version))
 
    era_row = conn.execute("SELECT * FROM roadmap_eras LIMIT 1").fetchone()
    _write(out_dir, f"eras/{era_row['era_id']}.md", rc.Era(
        schema_version=rc.SCHEMA_VERSION, id=era_row["era_id"],
        era_label=_one_line(era_row["era_label"]) or "roadmap era",
        started_at=_iso_date(era_row["started_at"]), order=era_row["sort_order"]))
 
    for trow in conn.execute("SELECT * FROM roadmap_tracks ORDER BY sort_order"):
        _write(out_dir, f"tracks/{trow['track_code']}.md", rc.Track(
            schema_version=rc.SCHEMA_VERSION, track_code=trow["track_code"],
            label=_one_line(trow["label"]) or trow["track_code"],
            description=_one_line(trow["description"]) or None,
            order=trow["sort_order"], active=bool(trow["active"])))
 
    for vrow in conn.execute("SELECT * FROM roadmap_versions ORDER BY version_id"):
        _write(out_dir, f"changelog/roadmap/{vrow['version_id']}.md", rc.Version(
            schema_version=rc.SCHEMA_VERSION, version=vrow["version_id"],
            date=_iso_date(vrow["released_at"]), era=vrow["era_id"], kind="roadmap",
            round=vrow["source_round_no"], notes=_one_line(vrow["notes"], 200) or None,
            entries=[rc.ChangelogEntry(scope="roadmap", text=f"{vrow['version_id']} seed import")]))
 
    _emit_phases(conn, out_dir, titles, deps, tracks, sanitized, report)
    _emit_items(conn, out_dir, deps, tracks, sanitized, report)
 
    counts = {t: conn.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0]
              for t in ("roadmap_eras", "roadmap_tracks", "roadmap_versions", "roadmap_phases", "roadmap_items_v3")}
    report.counts = counts
    conn.close()
    _dbg("OC_CHAT.MAIN", func, f"emitted source (promotions={len(report.promotions)})", 2)
    return report
 
 
def _sub_rows(conn: sqlite3.Connection, phase_code: str) -> list:
    return conn.execute(
        # OCT-LINT: disable=no_hardcoded_secrets reason="Seed migration constants + placeholder slot are not secrets"
        "SELECT * FROM roadmap_items_v3 WHERE parent_item_id = ? ORDER BY sort_order, item_id",
        (phase_code,),
    ).fetchall()
 
 
def _emit_phases(conn, out_dir, titles, deps, tracks, sanitized, report) -> None:
    phase_rows = conn.execute(
        # OCT-LINT: disable=no_hardcoded_secrets reason="Seed migration constants + placeholder slot are not secrets"
        "SELECT p.*, i.status FROM roadmap_phases p JOIN roadmap_items_v3 i ON i.item_id = p.phase_code "
        "WHERE i.parent_item_id IS NULL ORDER BY p.sort_order"
    ).fetchall()
    for prow in phase_rows:
        code = prow["phase_code"]
        parent_done = prow["status"] == "done"
        sub_models: list = []
        for srow in _sub_rows(conn, code):
            sid = sanitized.get(srow["item_id"], srow["item_id"])
            status = _reconcile_child_status(sid, srow["status"], parent_done, report)
            if _needs_promotion(srow, deps):
                report.promotions.append({"id": sid, "reason": _promotion_reason(srow, deps)})
                _write(out_dir, f"phases/{code}/{sid}.md", PromotedItem(
                    schema_version=rc.SCHEMA_VERSION, id=sid, title=_one_line(srow["title"], 240) or sid,
                    purpose=_derive_purpose(srow["title"], srow["prose"], report), status=status,
                    priority=_priority(srow["priority"]), order=srow["sort_order"],
                    depends_on=_depends_on(srow["item_id"], deps, sanitized)),
                    body=_one_line(srow["prose"], 100000))
            else:
                sub_models.append(rc.SubPhase(id=sid, status=status,
                                              description=_one_line(srow["title"], 240) or sid))
        priority = _priority(prow["priority_label"])
        if priority is None:
            priority = "lower"  # top-level phases all carry one; defensive
        title = titles.get(code) or _one_line(prow["label"]) or code
        _write(out_dir, f"phases/{code}.md", rc.Phase(
            schema_version=rc.SCHEMA_VERSION, id=code, era=conn.execute("SELECT era_id FROM roadmap_eras LIMIT 1").fetchone()[0],
            title=title, purpose=_one_line(title, 240), priority=priority, status=prow["status"],
            depends_on=_depends_on(code, deps, sanitized), track_codes=tracks.get(code, []),
            order=prow["sort_order"],
            sub_phases=sub_models))
 
 
def _emit_items(conn, out_dir, deps, tracks, sanitized, report) -> None:
    rows = conn.execute(
        # OCT-LINT: disable=no_hardcoded_secrets reason="Seed migration constants + placeholder slot are not secrets"
        "SELECT * FROM roadmap_items_v3 WHERE track_kind != 'm_phase' ORDER BY track_kind, sort_order, item_id"
    ).fetchall()
    for row in rows:
        iid = sanitized.get(row["item_id"], row["item_id"])
        _write(out_dir, f"items/{row['track_kind']}/{iid}.md", rc.RoadmapItem(
            schema_version=rc.SCHEMA_VERSION, id=iid, track_kind=row["track_kind"],
            title=_one_line(row["title"], 240) or iid,
            purpose=_derive_purpose(row["title"], row["prose"], report),
            status=row["status"], priority=_priority(row["priority"]),
            depends_on=_depends_on(row["item_id"], deps, sanitized),
            track_codes=tracks.get(row["item_id"], []), order=row["sort_order"]),
            body=_one_line(row["prose"], 100000))
 
 
def _reconcile_child_status(item_id: str, status: str, parent_done: bool, report: MigrationReport) -> str:
    """A done parent's non-terminal child is reconciled to ``deferred`` (it was
    deferred when the milestone shipped) — preserves the done milestone signal
    while satisfying the done-parent⇒children-terminal invariant (AC §8.50).
    Flagged for operator review (§9 3b)."""
    if parent_done and status not in rc.TERMINAL_STATUSES:
        report.reconciled_statuses.append({"id": item_id, "from": status, "to": "deferred"})
        return "deferred"
    return status
 
 
def _priority(value: object) -> object:
    return value if value in _PRIORITIES else None
 
 
def _promotion_reason(row: sqlite3.Row, deps: dict) -> str:
    reasons = []
    if len((row["title"] or "")) > 240:
        reasons.append("title>240")
    if row["priority"] is not None:
        reasons.append("priority")
    if deps.get(row["item_id"]):
        reasons.append("dependencies")
    return ",".join(reasons)