ingest.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/roadmap/ingest.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/roadmap/test_ingest_idempotence.py"
"""
Idempotent ROADMAP corpus ingest into the registry.
 
Purpose
-------
Populate the roadmap v0.3 registry tables from the on-disk ROADMAP corpus.
 
Responsibilities
----------------
- Walk and parse the four ROADMAP artifact families.
- Seed static version, phase, track, and render-section metadata.
- Emit roadmap item and dependency operations only for semantic changes.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - ingest lifecycle
    L3 - file and row counts
    L4 - per-record idempotence decisions
 
Contracts
---------
- Re-running against unchanged inputs creates no new operation_log rows.
- Dry-run reports pending changes without writing registry rows.
 
Dependencies
------------
- oc_chat.registry.roadmap.parser (markdown parsing)
- oc_chat.registry.roadmap.operations (mutating seam helpers)
- oc_chat.registry.api (Registry facade the operations are executed through)
- oc_chat.registry.backends.sqlite (SQLiteBackend connections and transactions for the metadata upserts)
- oc_chat.registry.canonical (canonical JSON for the idempotence hash)
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
from dataclasses import dataclass
from pathlib import Path
import sqlite3
from typing import Mapping
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.api import Registry
from oc_chat.registry.backends.sqlite import SQLiteBackend
from oc_chat.registry.canonical import canonical_json
from oc_chat.registry.roadmap.operations import (
    roadmap_item_add_dependency,
    roadmap_item_create,
)
from oc_chat.registry.roadmap.parser import (
    ParsedItem,
    ParsedPhase,
    ParsedRoadmap,
    ParsedRenderSection,
    ParsedTrack,
    parse_roadmap_file,
)
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
@dataclass(frozen=True)
class IngestResult:
    ok: bool
    dry_run: bool
    files: int
    created: int
    updated: int
    skipped: int
    errors: tuple[str, ...] = ()
    warnings: tuple[str, ...] = ()
 
    def to_dict(self) -> dict[str, object]:
        return {
            "ok": self.ok,
            "dry_run": self.dry_run,
            "files": self.files,
            "created": self.created,
            "updated": self.updated,
            "skipped": self.skipped,
            "errors": list(self.errors),
            "warnings": list(self.warnings),
        }
 
 
def ingest_corpus(
    registry: Registry,
    roadmap_dir: str | Path,
    *,
    actor_slot: str | None = None,
    actor_type: str = "human_operator",
    dry_run: bool = False,
) -> IngestResult:
    """Parse and ingest every ROADMAP artifact under *roadmap_dir*."""
 
    backend = _sqlite_backend(registry)
    slot = actor_slot or backend.active_operator_slot()
    if slot is None:
        return IngestResult(
            ok=False,
            dry_run=dry_run,
            files=0,
            created=0,
            updated=0,
            skipped=0,
            errors=("no active operator slot exists",),
        )
    path = Path(roadmap_dir)
    files = _roadmap_files(path)
    parsed: list[ParsedRoadmap] = []
    errors: list[str] = []
    for file_path in files:
        try:
            parsed.append(parse_roadmap_file(file_path))
        except Exception as exc:
            errors.append(f"{file_path}: {exc}")
    if errors:
        return IngestResult(
            ok=False,
            dry_run=dry_run,
            files=len(files),
            created=0,
            updated=0,
            skipped=0,
            errors=tuple(errors),
        )
 
    created = 0
    updated = 0
    skipped = 0
    if dry_run:
        with backend.connect() as conn:
            for roadmap in parsed:
                for item in roadmap.items:
                    action = _item_action(conn, item)
                    if action == "create":
                        created += 1
                    elif action == "update":
                        updated += 1
                    else:
                        skipped += 1
        return IngestResult(
            ok=True,
            dry_run=True,
            files=len(files),
            created=created,
            updated=updated,
            skipped=skipped,
        )
 
    with backend.transaction("immediate") as conn:
        _upsert_static_metadata(conn, parsed, backend.clock.now_wall())
 
    for roadmap in parsed:
        for item in roadmap.items:
            with backend.connect() as conn:
                action = _item_action(conn, item)
            if action == "skip":
                skipped += 1
                continue
            roadmap_item_create(
                registry,
                item_id=item.item_id,
                title=item.title,
                track_kind=item.track_kind,  # type: ignore[arg-type]
                phase_code=item.phase_code,
                parent_item_id=item.parent_item_id,
                prose=item.prose,
                status=item.status,  # type: ignore[arg-type]
                priority=item.priority,  # type: ignore[arg-type]
                sort_order=item.sort_order,
                target_version=item.target_version,
                introduced_in_version_id=roadmap.version_id,
                actor_slot=slot,
                actor_type=actor_type,  # type: ignore[arg-type]
                command="roadmap import",
                system_actor=True,
            )
            if action == "create":
                created += 1
            else:
                updated += 1
 
    with backend.transaction("immediate") as conn:
        _upsert_item_tracks(conn, parsed, backend.clock.now_wall())
 
    for roadmap in parsed:
        for dependency in roadmap.dependencies:
            with backend.connect() as conn:
                exists = conn.execute(
                    # OCT-LINT: disable=no_hardcoded_secrets reason="Roadmap SQL projection strings are schema constants, not secrets"
                    "SELECT 1 FROM roadmap_item_dependencies "
                    "WHERE from_item_id = ? AND to_item_id = ? AND kind = ?",
                    (
                        dependency.from_item_id,
                        dependency.to_item_id,
                        dependency.kind,
                    ),
                ).fetchone()
            if exists is not None:
                continue
            roadmap_item_add_dependency(
                registry,
                from_item_id=dependency.from_item_id,
                to_item_id=dependency.to_item_id,
                kind=dependency.kind,  # type: ignore[arg-type]
                in_version_id=roadmap.version_id,
                actor_slot=slot,
                actor_type=actor_type,  # type: ignore[arg-type]
                command="roadmap import",
                system_actor=True,
            )
 
    return IngestResult(
        ok=True,
        dry_run=False,
        files=len(files),
        created=created,
        updated=updated,
        skipped=skipped,
    )
 
 
def _roadmap_files(path: Path) -> list[Path]:
    if not path.is_dir():
        raise FileNotFoundError(path)
    return sorted(
        item
        for item in path.glob("ROADMAP*.md")
        if item.is_file() and "-v" in item.stem
    )
 
 
def _upsert_static_metadata(
    conn: sqlite3.Connection,
    roadmaps: list[ParsedRoadmap],
    now: str,
) -> None:
    latest_by_era: dict[str, str] = {}
    for roadmap in sorted(roadmaps, key=lambda item: item.version_id):
        _upsert_era(conn, roadmap, now)
        _upsert_version(conn, roadmap, "draft")
        latest_by_era[roadmap.era_id] = roadmap.version_id
        for track in roadmap.tracks:
            _upsert_track(conn, track)
        for phase in roadmap.phases:
            _upsert_phase(conn, phase, roadmap.version_id, now)
        for section in roadmap.render_sections:
            _upsert_render_section(conn, section, roadmap.version_id)
    for era_id, version_id in latest_by_era.items():
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Roadmap SQL projection strings are schema constants, not secrets"
            "UPDATE roadmap_versions SET status = 'draft' "
            "WHERE era_id = ? AND version_id != ? AND status = 'current'",
            (era_id, version_id),
        )
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Roadmap SQL projection strings are schema constants, not secrets"
            "UPDATE roadmap_versions SET status = 'current' WHERE version_id = ?",
            (version_id,),
        )
 
 
def _upsert_era(conn: sqlite3.Connection, roadmap: ParsedRoadmap, now: str) -> None:
    conn.execute(
        # OCT-LINT: disable=no_hardcoded_secrets reason="Roadmap SQL projection strings are schema constants, not secrets"
        "INSERT INTO roadmap_eras("
        "era_id, era_label, major_bump_reason, started_at, ended_at, archived_under, sort_order"
        ") VALUES (?, ?, ?, ?, NULL, NULL, ?) "
        "ON CONFLICT(era_id) DO UPDATE SET "
        "era_label = excluded.era_label",
        (
            roadmap.era_id,
            f"{roadmap.era_id} roadmap era",
            None if roadmap.era_id != "v3" else "Migration era",
            roadmap.created_at or now,
            _era_sort_order(roadmap.era_id),
        ),
    )
 
 
def _upsert_version(
    conn: sqlite3.Connection,
    roadmap: ParsedRoadmap,
    status: str,
) -> None:
    conn.execute(
        # OCT-LINT: disable=no_hardcoded_secrets reason="Roadmap SQL projection strings are schema constants, not secrets"
        "INSERT INTO roadmap_versions("
        "version_id, era_id, released_at, status, parent_version_id, "
        "supersedes_version_id, superseded_by_version_id, last_updated_at, "
        "source_round_no, bump_kind, notes"
        ") VALUES (?, ?, ?, ?, NULL, NULL, NULL, ?, NULL, ?, ?) "
        "ON CONFLICT(version_id) DO UPDATE SET "
        "released_at = excluded.released_at, "
        "last_updated_at = excluded.last_updated_at, "
        "bump_kind = excluded.bump_kind, "
        "notes = excluded.notes",
        (
            roadmap.version_id,
            roadmap.era_id,
            roadmap.created_at,
            status,
            roadmap.last_updated_at,
            "patch",
            roadmap.title,
        ),
    )
 
 
def _upsert_track(conn: sqlite3.Connection, track: ParsedTrack) -> None:
    conn.execute(
        # OCT-LINT: disable=no_hardcoded_secrets reason="Roadmap SQL projection strings are schema constants, not secrets"
        "INSERT INTO roadmap_tracks(track_code, label, description, sort_order, active) "
        "VALUES (?, ?, ?, ?, 1) "
        "ON CONFLICT(track_code) DO UPDATE SET "
        "label = excluded.label, "
        "description = excluded.description, "
        "sort_order = excluded.sort_order, "
        "active = 1",
        (track.track_code, track.label, track.description, track.sort_order),
    )
 
 
def _upsert_phase(
    conn: sqlite3.Connection,
    phase: ParsedPhase,
    version_id: str,
    now: str,
) -> None:
    parent_id = None
    if phase.parent_phase_code is not None:
        row = conn.execute(
            "SELECT phase_id FROM roadmap_phases WHERE phase_code = ?",
            (phase.parent_phase_code,),
        ).fetchone()
        if row is not None:
            parent_id = int(row["phase_id"])
    conn.execute(
        # OCT-LINT: disable=no_hardcoded_secrets reason="Roadmap SQL projection strings are schema constants, not secrets"
        "INSERT INTO roadmap_phases("
        "phase_code, label, parent_phase_id, branch_suggestion, commit_prefix, "
        "target_version, priority_label, parallel_track, full_spec_url, sort_order, "
        "introduced_in_version_id, retired_in_version_id, created_at, updated_at"
        ") VALUES (?, ?, ?, NULL, NULL, ?, ?, ?, NULL, ?, ?, NULL, ?, ?) "
        "ON CONFLICT(phase_code) DO UPDATE SET "
        "label = excluded.label, "
        "parent_phase_id = COALESCE(roadmap_phases.parent_phase_id, excluded.parent_phase_id), "
        "priority_label = COALESCE(excluded.priority_label, roadmap_phases.priority_label), "
        "parallel_track = excluded.parallel_track, "
        "sort_order = excluded.sort_order, "
        "updated_at = excluded.updated_at",
        (
            phase.phase_code,
            phase.label,
            parent_id,
            phase.target_version,
            phase.priority_label,
            1 if phase.parallel_track else 0,
            phase.sort_order,
            version_id,
            now,
            now,
        ),
    )
 
 
def _upsert_render_section(
    conn: sqlite3.Connection,
    section: ParsedRenderSection,
    version_id: str,
) -> None:
    conn.execute(
        # OCT-LINT: disable=no_hardcoded_secrets reason="Roadmap SQL projection strings are schema constants, not secrets"
        "DELETE FROM roadmap_render_sections WHERE file_kind = ? AND section_path = ?",
        (section.file_kind, section.section_path),
    )
    conn.execute(
        # OCT-LINT: disable=no_hardcoded_secrets reason="Roadmap SQL projection strings are schema constants, not secrets"
        "INSERT INTO roadmap_render_sections("
        "file_kind, section_path, sort_order, query_kind, query_params_json, "
        "required_min_version, active"
        ") VALUES (?, ?, ?, ?, ?, ?, 1)",
        (
            section.file_kind,
            section.section_path,
            section.sort_order,
            section.query_kind,
            section.query_params_json,
            version_id,
        ),
    )
 
 
def _upsert_item_tracks(
    conn: sqlite3.Connection,
    roadmaps: list[ParsedRoadmap],
    now: str,
) -> None:
    for roadmap in roadmaps:
        for item in roadmap.items:
            track_code = _track_code_for_item(item)
            row = conn.execute(
                "SELECT track_id FROM roadmap_tracks WHERE track_code = ?",
                (track_code,),
            ).fetchone()
            if row is None:
                continue
            conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Roadmap SQL projection strings are schema constants, not secrets"
                "INSERT OR IGNORE INTO roadmap_item_tracks(item_id, track_id, created_at) "
                "VALUES (?, ?, ?)",
                (item.item_id, int(row["track_id"]), now),
            )
 
 
def _item_action(conn: sqlite3.Connection, item: ParsedItem) -> str:
    row = conn.execute(
        # OCT-LINT: disable=no_hardcoded_secrets reason="Roadmap SQL projection strings are schema constants, not secrets"
        "SELECT i.item_id, i.track_kind, p.phase_code, i.parent_item_id, "
        "i.title, i.prose, i.status, i.priority, i.sort_order, i.target_version "
        "FROM roadmap_items_v3 i "
        "LEFT JOIN roadmap_phases p ON p.phase_id = i.phase_id "
        "WHERE i.item_id = ?",
        (item.item_id,),
    ).fetchone()
    if row is None:
        return "create"
    current = {
        "track_kind": row["track_kind"],
        "phase_code": row["phase_code"],
        "parent_item_id": row["parent_item_id"],
        "title": row["title"],
        "prose": row["prose"],
        "status": row["status"],
        "priority": row["priority"],
        "sort_order": int(row["sort_order"]),
        "target_version": row["target_version"],
    }
    incoming = {
        "track_kind": item.track_kind,
        "phase_code": item.phase_code,
        "parent_item_id": item.parent_item_id,
        "title": item.title,
        "prose": item.prose,
        "status": item.status,
        "priority": item.priority,
        "sort_order": item.sort_order,
        "target_version": item.target_version,
    }
    return "skip" if _canonical_hash(current) == _canonical_hash(incoming) else "update"
 
 
def _canonical_hash(value: Mapping[str, object]) -> str:
    import hashlib
 
    return hashlib.sha256(canonical_json(dict(value)).encode("utf-8")).hexdigest()
 
 
def _track_code_for_item(item: ParsedItem) -> str:
    if item.track_kind in {"m_phase", "vision", "git_workflow", "pre_m0_era"}:
        return item.track_kind
    if item.item_id.startswith("L"):
        return "memory"
    if item.item_id.startswith("U"):
        return "undo"
    return "tooling"
 
 
def _era_sort_order(era_id: str) -> int:
    text = era_id.removeprefix("v")
    return int(text) if text.isdigit() else 999
 
 
def _sqlite_backend(registry: Registry) -> SQLiteBackend:
    backend = registry.backend
    if not isinstance(backend, SQLiteBackend):
        raise TypeError("roadmap ingest requires SQLiteBackend")
    return backend