engine.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/roadmap/reconcile/engine.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/roadmap/test_reconcile.py"
"""
Reconciliation run + plan-link persistence (Plan 205).
 
Purpose
-------
Compose the three check classes into one report, and persist the plan-to-item
join into ``roadmap_plan_links`` — the table designed for it, unused since it
was created.
 
Responsibilities
----------------
- ``reconcile`` — pure composition over injected indexes and history.
- ``reconcile_workspace`` — the convenience wrapper that loads a real workspace.
- ``write_plan_links`` — persist the join, after a preflight that names every
  missing foreign-key parent instead of fabricating one.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - run lifecycle
    L3 - per-class and per-write summaries
    L4 - per-row preflight details
 
Contracts
---------
- ``reconcile`` performs no writes of any kind.
- ``write_plan_links`` targets a **live** registry, never the authored
  projection: the projection is rebuilt into a temp store and swapped on every
  ``regenerate``, so a row written there is destroyed by the next round close,
  and ``roadmap_plan_links`` is a runtime-excluded table by design.
- The writer never creates a foreign-key parent to make its own insert succeed.
  A missing ``plans`` or ``roadmap_items_v3`` row is reported, not invented:
  the check reports, the operator remediates.
- ``roadmap_item_events`` is deliberately NOT written. The only operation that
  appends one, ``roadmap_item.set_status``, also mutates the item's status, and
  a reconciliation pass must never change the state it is auditing.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.api (Registry facade)
- oc_chat.registry.roadmap.operations (the plan-link operation executor)
- oc_chat.registry.roadmap.reconcile.checks (the three check classes)
- oc_chat.registry.roadmap.reconcile.history (the git history reader for class 1)
- oc_chat.registry.roadmap.reconcile.inputs (the source index and plan record loaders)
- oc_chat.registry.roadmap.reconcile.model (the finding, link, transition and report types)
"""
from __future__ import annotations
 
from dataclasses import dataclass
from pathlib import Path
from typing import Sequence
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.api import Registry
from oc_chat.registry.roadmap.operations import ActorType, roadmap_plan_link_create
from oc_chat.registry.roadmap.reconcile.checks import (
    check_advances_conformance,
    check_changelog_status_divergence,
    check_stale_completion,
)
from oc_chat.registry.roadmap.reconcile.history import (
    CommitRecord,
    HistoryError,
    read_history,
)
from oc_chat.registry.roadmap.reconcile.inputs import (
    PlanRecord,
    SourceIndex,
    load_plan_records,
    load_source_index,
)
from oc_chat.registry.roadmap.reconcile.model import (
    Finding,
    PlanLink,
    ReconcileReport,
    StatusTransition,
    count_by_kind,
)
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
#: Workspace-relative default locations.
DEFAULT_SOURCE_REL = "oc_chat/oc_chat/roadmap/source"
DEFAULT_PLANS_REL = "oc_chat/oc_chat/plans/completed"
 
_SEVERITY_ORDER = {"violation": 0, "advisory": 1}
 
 
class ReconcileConfigError(ValueError):
    """The check was pointed at something it cannot read as a result.
 
    A ``--source`` outside ``--repo`` yields a git pathspec that can never match,
    which would otherwise surface as "0 records, nothing wrong". A pathspec that
    cannot match is a configuration error, not a clean bill of health.
    """
 
 
def _resolve_scope(root: Path, source: Path) -> str:
    """Return the repo-relative POSIX pathspec for ``source``, or refuse.
 
    Both paths are **resolved first**. The original bug (R1) asked
    ``Path.is_relative_to`` about an unresolved, cwd-relative source, decided it
    lay outside the repo, and passed the raw relative string to ``git log`` as a
    pathspec — where it silently matched nothing.
    """
 
    if not source.is_relative_to(root):
        raise ReconcileConfigError(
            f"--source is not inside --repo, so no git pathspec can match it.\n"
            f"  --source resolved to: {source}\n"
            f"  --repo   resolved to: {root}\n"
            "Point --source at a path inside the repository, or pass the matching "
            "--repo. Refusing to report a result for a scope that cannot be read."
        )
    return source.relative_to(root).as_posix()
 
 
def _sort_key(finding: Finding) -> tuple[int, str, str]:
    """Order findings violations-first, then by kind, then by subject."""
 
    return (_SEVERITY_ORDER.get(finding.severity, 9), finding.kind, finding.subject)
 
 
def _scope_blind_spot(
    records_examined: int, records_on_disk: int, history_error: str | None
) -> str | None:
    """Flag class 1 as not-evaluated when it examined nothing it could have.
 
    This is the backstop for every cause not named by ``_resolve_scope``: a
    ``--since`` window matching no commits, a corpus whose records were never
    committed, a pathspec typo that still lands inside the repo. If there was
    something on disk to examine and nothing was examined, the class verified
    NOTHING — which must never render as a clean zero.
 
    An empty changelog corpus is left evaluated: nothing to examine is a real
    result, and crying wolf there would train readers to ignore the signal.
    """
 
    if history_error is not None or records_examined > 0 or records_on_disk == 0:
        return None
    return (
        f"class 1 examined 0 changelog records, but {records_on_disk} exist on disk "
        "under the resolved source tree — the commit window or source pathspec "
        "matched nothing, so this class verified NOTHING (it is not clean)"
    )
 
 
def reconcile(
    *,
    source_index: SourceIndex,
    plan_records: Sequence[PlanRecord] = (),
    history: Sequence[CommitRecord] = (),
    repo_root: Path | None = None,
    advances_required_from: str | None = None,
    history_error: str | None = None,
) -> ReconcileReport:
    """Run all three check classes over the supplied inputs.
 
    Pure: every input is injected, so the engine is exercised without git and
    without a registry.
    """
 
    func = "reconcile"
    findings: list[Finding] = []
    findings += check_changelog_status_divergence(source_index, history)
    advances_findings, links, none_count = check_advances_conformance(
        source_index, plan_records, advances_required_from=advances_required_from
    )
    findings += advances_findings
    findings += check_stale_completion(source_index, repo_root=repo_root)
    findings.sort(key=_sort_key)
 
    transitions = tuple(
        StatusTransition(
            record_id=change.record_id,
            from_status=change.from_status,
            to_status=change.to_status,
            sha=commit.sha,
            date=commit.date,
        )
        for commit in history
        for change in commit.status_changes
    )
 
    records_examined = sum(len(commit.added_changelog) for commit in history)
    report = ReconcileReport(
        findings=tuple(findings),
        links=tuple(links),
        transitions=transitions,
        counts=count_by_kind(findings),
        advances_none_count=none_count,
        plans_checked=len(plan_records),
        changelog_records_checked=records_examined,
        records_checked=len(source_index.records),
        unscannable_ids=source_index.unscannable_ids,
        history_error=history_error,
        changelog_scope_error=_scope_blind_spot(
            records_examined, len(source_index.changelog), history_error
        ),
    )
    _dbg(
        "OC_CHAT.MAIN",
        func,
        f"{report.violation_count} violations / {report.advisory_count} advisories",
        2,
    )
    return report
 
 
def reconcile_workspace(
    workspace_root: str | Path,
    *,
    source_dir: str | Path | None = None,
    plans_dir: str | Path | None = None,
    since: str | None = None,
    advances_required_from: str | None = None,
) -> ReconcileReport:
    """Load a real workspace and reconcile it."""
 
    func = "reconcile_workspace"
    root = Path(workspace_root).resolve()
    source = (Path(source_dir) if source_dir else root / DEFAULT_SOURCE_REL).resolve()
    plans = (Path(plans_dir) if plans_dir else root / DEFAULT_PLANS_REL).resolve()
    source_rel = _resolve_scope(root, source)
    _dbg("OC_CHAT.MAIN", func, f"reconciling {root} (scope {source_rel})", 2)
    # Class 1 is the only class that needs git. When history is unreadable the
    # other two still run, but the report says so explicitly: a zero class-1
    # count must never be mistaken for a clean one.
    history: tuple[CommitRecord, ...] = ()
    history_error: str | None = None
    try:
        history = read_history(root, source_rel=source_rel, since=since)
    except HistoryError as error:
        history_error = str(error)
        _dbg("OC_CHAT.MAIN", func, "class 1 not evaluated: no git history", 2)
    return reconcile(
        source_index=load_source_index(source),
        plan_records=load_plan_records(plans),
        history=history,
        repo_root=root,
        advances_required_from=advances_required_from,
        history_error=history_error,
    )
 
 
@dataclass(frozen=True)
class JoinWriteResult:
    """The outcome of persisting the plan-link join."""
 
    ok: bool
    links_written: int
    missing_plans: tuple[str, ...]
    missing_items: tuple[str, ...]
    detail: str
 
    def to_dict(self) -> dict[str, object]:
        """The JSON projection of this write attempt."""
 
        return {
            "ok": self.ok,
            "links_written": self.links_written,
            "missing_plans": list(self.missing_plans),
            "missing_items": list(self.missing_items),
            "detail": self.detail,
        }
 
 
def _existing(registry: Registry, table: str, column: str, wanted: set[str]) -> set[str]:
    """The subset of ``wanted`` that already exists in ``table``.``column``."""
 
    if not wanted:
        return set()
    backend = registry.backend
    with backend.connect() as conn:  # type: ignore[attr-defined]
        placeholders = ",".join("?" for _ in wanted)
        rows = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Registry table and column names are schema constants, not secrets"
            f"SELECT {column} FROM {table} WHERE {column} IN ({placeholders})",
            tuple(sorted(wanted)),
        ).fetchall()
    return {str(row[0]) for row in rows}
 
 
def write_plan_links(
    registry: Registry,
    links: Sequence[PlanLink],
    *,
    actor_slot: str,
    actor_type: ActorType = "human_operator",
    system_actor: bool = True,
    dry_run: bool = False,
) -> JoinWriteResult:
    """Persist ``links`` into ``roadmap_plan_links`` after a fail-loud preflight.
 
    The two foreign-key parents of this table — ``plans(plan_id)`` and
    ``roadmap_items_v3(item_id)`` — must both be present in the target store.
    When either is missing the write is refused whole and the missing ids are
    named; nothing is fabricated to make the insert succeed.
 
    ``actor_type`` defaults to ``human_operator`` because the CLI resolves the
    store's active *operator* slot, and the backend rejects a slot whose
    registered type disagrees with the claimed one.
    """
 
    func = "write_plan_links"
    if not links:
        return JoinWriteResult(True, 0, (), (), "no links to write")
 
    have_plans = _existing(registry, "plans", "plan_id", {link.plan_id for link in links})
    have_items = _existing(
        registry, "roadmap_items_v3", "item_id", {link.item_id for link in links}
    )
    missing_plans = tuple(sorted({link.plan_id for link in links} - have_plans))
    missing_items = tuple(sorted({link.item_id for link in links} - have_items))
    if missing_plans or missing_items:
        detail = (
            "refused: roadmap_plan_links has two foreign-key parents and this store "
            f"is missing {len(missing_plans)} plan row(s) and {len(missing_items)} "
            "roadmap item row(s). The check reports; it does not invent parents. "
            "Populate the live store first (queue the plans; `roadmap import` the "
            "items) or resolve where the join is meant to live."
        )
        _dbg("OC_CHAT.MAIN", func, "preflight refused the write", 2)
        return JoinWriteResult(False, 0, missing_plans, missing_items, detail)
 
    if dry_run:
        return JoinWriteResult(True, 0, (), (), f"dry run: {len(links)} link(s) would be written")
 
    for link in links:
        roadmap_plan_link_create(
            registry,
            plan_id=link.plan_id,
            item_id=link.item_id,
            contribution=link.contribution,
            actor_slot=actor_slot,
            actor_type=actor_type,
            command="roadmap reconcile --write-links",
            system_actor=system_actor,
        )
    _dbg("OC_CHAT.MAIN", func, f"wrote {len(links)} plan links", 2)
    return JoinWriteResult(True, len(links), (), (), f"wrote {len(links)} plan link(s)")