pipeline.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/regenerate/pipeline.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/regenerate/test_pipeline.py"
"""
The 7-step ``regenerate`` orchestration (Plan 142 P1.4, §5.3).
 
Purpose
-------
Drive scan -> parse/load -> local schema -> canonicalisation-check ->
``source_hash`` -> cross-record -> temp-DB build + derive-verify +
verification-stamp publish-gate + WAL-safe atomic swap, returning a result that
carries the §10 envelope. A later step never begins until the earlier one
passes; a failure carries the §10 hash-field nullability for its stage.
 
Responsibilities
----------------
- ``regenerate(source_dir, db_path, ...)`` — run the pipeline, build the
  worktree-local store, never touching the shared main-repo registry.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - pipeline lifecycle
    L3 - per-step progress
    L4 - swap/verify details
 
Contracts
---------
- Idempotent: identical inputs -> identical source/logical hashes (no
  hash-contributing row changes on a no-op).
- Verify-before-USE: the built DB is never swapped/hashed/read until the
  derive-verify passes; the stamp is written before the atomic swap.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.backends.sqlite (SQLiteBackend and RegistrySeed to initialise the temp DB)
- oc_chat.registry.regenerate.build (build_roadmap inserts the roadmap rows into the temp DB)
- oc_chat.registry.regenerate.cross_record (validate_cross_record, the step-6 cross-record checks)
- oc_chat.registry.regenerate.derive_verify (version dedup, derivation and the step-7 derive-verify gate)
- oc_chat.registry.regenerate.envelope (RegenerateError, the REGEN_* and SRC_STRICT_TYPE codes and build_envelope)
- oc_chat.registry.regenerate.lock (acquire and release the regenerate lock around the run)
- oc_chat.registry.regenerate.verified_projection (write and assert the verification stamp before the swap)
- oc_chat.registry.source.canonical (record loading, frontmatter split, YAML feature guard and the canonicalisation check)
- oc_chat.registry.source.errors (SourceError caught for the failure envelope, SourceValidationError raised on a promoted item's strict-type failure and SourceScanError to tell the scan stage apart)
- oc_chat.registry.source.hashing (raw-source-tree, source and logical-state hashes)
- oc_chat.registry.source.promoted_item (PromotedItem model for promoted item records)
- oc_chat.registry.source.scanner (scan_source_tree, step 1 of the pipeline)
- oc_chat.registry.store_scope (stamp_authored_projection so live commands refuse the built store)
- pydantic (ValidationError from PromotedItem validation)
- yaml (safe_load of promoted item frontmatter)
"""
from __future__ import annotations
 
import gc
import hashlib
import os
import sqlite3
import uuid
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Optional
 
import yaml
from pydantic import ValidationError
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.backends.sqlite import RegistrySeed, SQLiteBackend
from oc_chat.registry.regenerate.build import build_roadmap
from oc_chat.registry.regenerate.cross_record import validate_cross_record
from oc_chat.registry.regenerate.derive_verify import (
    derive_versions,
    dedup_version_records,
    verify_versions,
)
from oc_chat.registry.regenerate import lock
from oc_chat.registry.regenerate.envelope import (
    REGEN_BUILD_FAILED,
    REGEN_SCOPE_CONFLICT,
    REGEN_SOURCE_MUTATED,
    REGEN_UNSUPPORTED_BACKEND,
    SRC_STRICT_TYPE,
    RegenerateError,
    build_envelope,
)
from oc_chat.registry.regenerate.verified_projection import (
    assert_verified_projection,
    write_verification_stamp,
)
from oc_chat.registry.source.canonical import (
    _assert_no_yaml_features,
    format_source_tree,
    load_record,
    split_record_text,
)
from oc_chat.registry.source.errors import SourceError, SourceValidationError
from oc_chat.registry.source.hashing import (
    logical_state_hash,
    raw_source_tree_hash,
    source_hash,
)
from oc_chat.registry.source.scanner import scan_source_tree
from oc_chat.registry.source.promoted_item import PromotedItem
from oc_chat.registry.store_scope import stamp_authored_projection
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
#: Runtime-only tables whose FK violations are tolerated on an authored rebuild
#: (Option A, §5.3 — the references are live-store integrity, not source).
_RUNTIME_TABLES = frozenset({"agents", "rounds", "plans", "operation_log"})
_ROADMAP_TABLES = (
    "roadmap_render_sections", "roadmap_item_dependencies", "roadmap_item_tracks",
    "roadmap_items_v3", "roadmap_phases", "roadmap_tracks", "roadmap_versions",
    "roadmap_eras",
)
 
 
@dataclass
class RegenerateResult:
    ok: bool
    envelope: dict[str, Any]
    db_path: Optional[Path] = None
    source_hash: Optional[str] = None
    logical_state_hash: Optional[str] = None
    semantic_hash: Optional[str] = None
    records: int = 0
    #: R5: the canonical surface for lock reclamations this run (Codex-v6-P2 /
    #: Fable-H9) — empty on the common no-reclaim path.
    lock_events: list["lock.LockReclaimEvent"] = field(default_factory=list)
    error: Optional[BaseException] = field(default=None, repr=False)
 
 
def _load_promoted(scanned: Any) -> tuple[PromotedItem, str]:
    raw = scanned.path.read_text(encoding="utf-8")
    frontmatter, body = split_record_text(raw)
    # [OP-5] parse-safety parity: run the §3.2 YAML event-scan on the promoted
    # load path exactly as the top-level ``load_record`` does — no bare
    # ``safe_load`` (a forbidden alias/anchor/tag in a promoted file is rejected).
    _assert_no_yaml_features(frontmatter, scanned.rel_path)
    data = yaml.safe_load(frontmatter) or {}
    try:
        model = PromotedItem.model_validate(data)
    except ValidationError as error:
        raise SourceValidationError(
            f"{scanned.rel_path}: promoted-item validation failed: {error}",
            code=SRC_STRICT_TYPE,
            records=[scanned.rel_path],
        ) from error
    if model.id != scanned.stem:
        raise SourceValidationError(
            f"{scanned.rel_path}: stem {scanned.stem!r} != id {model.id!r}",
            code=SRC_STRICT_TYPE,
            records=[scanned.rel_path],
        )
    return model, body
 
 
def _categorize(records: list) -> dict[str, list]:
    categorized: dict[str, Any] = {
        "manifest": None, "eras": [], "tracks": [], "versions": [],
        "phases": [], "items": [], "promoted": [], "framing": [], "notes": [],
    }
    for scanned in records:
        if scanned.kind == "promoted_item":
            model, body = _load_promoted(scanned)
            categorized["promoted"].append((scanned, model, body))
            continue
        model, body = load_record(scanned)
        if scanned.kind == "manifest":
            categorized["manifest"] = model
        elif scanned.kind == "era":
            categorized["eras"].append(model)
        elif scanned.kind == "track":
            categorized["tracks"].append(model)
        elif scanned.kind == "version":
            categorized["versions"].append(model)
        elif scanned.kind == "framing":
            # [NEW·V24] source-only projection (no store rows); validated (anchors)
            # in cross-record + rendered as generated framing (§3.3 / §6).
            categorized["framing"].append(model)
        elif scanned.kind == "phase":
            categorized["phases"].append((model, body))
        elif scanned.kind == "roadmap_item":
            categorized["items"].append((model, body))
        elif scanned.kind == "note":
            categorized["notes"].append((model, body))
    return categorized
 
 
def _init_temp_db(temp_path: Path) -> None:
    backend = SQLiteBackend(temp_path)
    backend.initialize(
        RegistrySeed(
            workspace_id=str(uuid.uuid4()),
            registry_id=str(uuid.uuid4()),
            workspace_root=str(temp_path.parent),
            git_common_dir=str(temp_path.parent / ".git"),
            registry_path=str(temp_path),
            checkpoint_path=str(temp_path.parent / ".option_c" / "checkpoint.json"),
            operator_slot="operator-source",
            hmac_key_id=str(uuid.uuid4()),
            secret_fingerprint=hashlib.sha256(b"authored-projection").hexdigest(),
            app_version="regenerate",
        )
    )
 
 
def _clear_roadmap_tables(conn: sqlite3.Connection) -> None:
    for table in _ROADMAP_TABLES:
        conn.execute(f"DELETE FROM {table}")
 
 
def _assert_source_foreign_keys(conn: sqlite3.Connection) -> None:
    """Run foreign_key_check, tolerating Option A runtime-table references."""
    for table, _rowid, referenced, _fkid in conn.execute("PRAGMA foreign_key_check").fetchall():
        if referenced in _RUNTIME_TABLES:
            continue  # Option A: runtime FKs not enforced on an authored rebuild
        # @invariant: inv.intra_source_fk_resolves  (§5.3.1 R — AC §8.32; the intra-source FK
        # resolution backstop. Option A tolerates runtime-table FKs (the X rows) above.)
        raise RegenerateError(
            f"foreign_key_check failed on {table} -> {referenced}",
            code=REGEN_BUILD_FAILED,
            records=[str(table)],
        )
 
 
def _build_and_publish(
    categorized: dict[str, list],
    db_path: Path,
    computed_source_hash: str,
    source_dir: Path,
    expected_raw_hash: str,
) -> str:
    """Step 7: build temp DB -> derive-verify -> stamp -> hash -> WAL-safe swap."""
    # A same-version roadmap+product changelog pair (agreement-validated in step 6)
    # collapses to one roadmap_versions row before derive/build/verify (§3.2 / AC §8.38).
    categorized = {**categorized, "versions": dedup_version_records(categorized.get("versions", []))}
    temp_path = db_path.with_name(db_path.name + ".regen-tmp.sqlite")
    for suffix in ("", "-wal", "-shm"):
        _best_effort_unlink(str(temp_path) + suffix)
    _init_temp_db(temp_path)
    derived = derive_versions(categorized.get("versions", []))
    conn = sqlite3.connect(str(temp_path))
    try:
        conn.execute("PRAGMA foreign_keys = OFF")
        _clear_roadmap_tables(conn)
        stamp_authored_projection(conn)  # refuse live commands (§5.3.1 / AC §8.57)
        build_roadmap(conn, categorized, derived)
        verify_versions(conn, categorized.get("versions", []))  # publish-gate
        computed_logical_hash = logical_state_hash(conn)
        write_verification_stamp(
            conn, source_hash=computed_source_hash, logical_state_hash=computed_logical_hash
        )
        assert_verified_projection(conn)  # must pass before swap
        # OCT-LINT: disable=no_hardcoded_secrets reason="Placeholder seed values + table names are not secrets"
        conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
    finally:
        conn.close()
    gc.collect()
    # R3 (§5.3 validate-before-swap): run integrity_check + foreign_key_check on
    # the TEMP DB *before* os.replace, so a corrupt/inconsistent build never
    # lands over the prior good DB. On failure we raise and the swap never runs,
    # leaving the prior DB + stamps intact.
    verify_temp = sqlite3.connect(str(temp_path))
    try:
        if verify_temp.execute("PRAGMA integrity_check").fetchone()[0] != "ok":
            raise RegenerateError(
                "integrity_check failed on temp DB before swap", code=REGEN_BUILD_FAILED
            )
        _assert_source_foreign_keys(verify_temp)
    finally:
        verify_temp.close()
    gc.collect()
    # Best-effort sidecar cleanup (§5.3: tolerate Windows transient locks post-checkpoint).
    for suffix in ("-wal", "-shm"):
        _best_effort_unlink(str(db_path) + suffix)
        _best_effort_unlink(str(temp_path) + suffix)
    # R5 pre-swap TOCTOU re-check (§5.3): re-hash the raw source tree and abort
    # *before* the swap if source/ mutated since the step-1 scan — preserving the
    # prior DB + stamps. The raw re-hash necessarily re-walks the file bytes; what
    # is forbidden (N-b) is a second scan/parse pass, not this byte re-hash. Pinned
    # to raw_source_tree_hash (not source_hash/semantic) so a canonical-equivalent
    # byte edit or a newly-added invalid file is caught as the retryable
    # REGEN_SOURCE_MUTATED, not an arbitrary source-validation error.
    if raw_source_tree_hash(source_dir) != expected_raw_hash:
        raise RegenerateError(
            "source/ mutated during regenerate (raw source-tree hash changed "
            "since the pre-build scan)",
            code=REGEN_SOURCE_MUTATED,
            hint="re-run regenerate on a quiescent source tree",
        )
    os.replace(str(temp_path), str(db_path))
    # Re-open post-swap is confirmation-only now (the fail-closed gate ran above).
    verify = sqlite3.connect(str(db_path))
    try:
        if verify.execute("PRAGMA integrity_check").fetchone()[0] != "ok":
            raise RegenerateError("integrity_check failed after swap", code=REGEN_BUILD_FAILED)
    finally:
        verify.close()
    gc.collect()
    return computed_logical_hash
 
 
def _best_effort_unlink(path: str) -> None:
    """Unlink ``path`` tolerating Windows transient locks (§5.3, post-checkpoint)."""
    try:
        Path(path).unlink(missing_ok=True)
    except OSError:
        pass
 
 
def regenerate(
    source_dir: Path,
    db_path: Path,
    *,
    backend: str = "sqlite",
    scope: str = "authored",
    lock_path: Optional[Path] = None,
    force: bool = False,
) -> RegenerateResult:
    """Run the 7-step pipeline; build the worktree-local store at ``db_path``.
 
    ``lock_path`` (R5): the regenerate lock file. The CLI (and any caller wanting
    same-worktree mutual exclusion) MUST pass the explicit workspace lock path
    (``<workspace>/.option_c/orchestration/regenerate.lock``, independent of
    ``--db``). ``lock_path=None`` is a **test/library compatibility default only**
    — resolved by the side-effect-free rule ``db_path.parent / "regenerate.lock"``
    (never source_dir-derived, never workspace discovery); it must never silently
    become the production contract (Codex-R04-P1). ``force`` reclaims a stale lock
    (logged as a ``LockReclaimEvent``; NEVER skips validation/derive-verify).
    """
    func = "regenerate"
    source_dir = Path(source_dir)
    db_path = Path(db_path)
    raw_hash: Optional[str] = None
    src_hash: Optional[str] = None
    if backend != "sqlite":
        error = RegenerateError(
            f"unsupported backend {backend!r}; v1 ships only sqlite",
            code=REGEN_UNSUPPORTED_BACKEND,
        )
        return RegenerateResult(False, build_envelope(error), error=error)
    if scope != "authored":
        # R4 (§5.3 fail-loud, operator 2026-07-05): --scope runtime|all is no
        # longer silently accepted-and-ignored — refuse with the typed code
        # (exit 2 via _EXIT_BY_CODE, remapped from the vacated class 4).
        error = RegenerateError(
            f"--scope {scope!r} is not supported; regenerate rebuilds only the "
            "authored projection",
            code=REGEN_SCOPE_CONFLICT,
            hint="drop --scope or pass --scope authored",
        )
        return RegenerateResult(False, build_envelope(error), error=error)
    resolved_lock = Path(lock_path) if lock_path is not None else db_path.parent / "regenerate.lock"
    try:
        handle = lock.acquire(resolved_lock, force=force)
    except RegenerateError as error:  # REGEN_LOCK_HELD (live-or-inconclusive holder)
        return RegenerateResult(False, build_envelope(error), error=error)
    lock_events = [handle.reclaim_event] if handle.reclaim_event is not None else []
    try:
        # Steps 1-5 (scan -> parse -> local -> canon-check -> source_hash).
        records = scan_source_tree(source_dir)
        raw_hash = raw_source_tree_hash(source_dir)
        categorized = _categorize(records)
        format_source_tree(records, check=True)  # step 4 canonicalisation check
        src_hash = source_hash(records)
        # Step 6 cross-record.
        validate_cross_record(categorized)
        # Step 7 build + derive-verify + stamp + pre-swap TOCTOU re-check + swap.
        logical_hash = _build_and_publish(categorized, db_path, src_hash, source_dir, raw_hash)
    except (SourceError, RegenerateError) as error:
        stage = _stage_for(error)
        envelope = build_envelope(
            error,
            raw_source_tree_hash=None if stage == "scan" else raw_hash,
            source_hash=src_hash if stage in ("cross_record", "build") else None,
        )
        return RegenerateResult(False, envelope, error=error, lock_events=lock_events)
    finally:
        lock.release(handle)
    _dbg("OC_CHAT.MAIN", func, "regenerate ok", 2)
    envelope = build_envelope(
        None, raw_source_tree_hash=raw_hash, source_hash=src_hash,
        logical_state_hash=logical_hash,
    )
    return RegenerateResult(
        True, envelope, db_path=db_path, source_hash=src_hash,
        logical_state_hash=logical_hash, records=len(records),
        lock_events=lock_events,
    )
 
 
def _stage_for(error: BaseException) -> str:
    from oc_chat.registry.source.errors import SourceScanError
 
    if isinstance(error, SourceScanError):
        return "scan"
    if isinstance(error, RegenerateError):
        return "build"
    return "cross_record"