inputs.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/roadmap/reconcile/inputs.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/roadmap/test_reconcile.py"
"""
Input readers for the roadmap reconciliation check (Plan 205).
 
Purpose
-------
Project the two authored corpora the check reads — the ``roadmap/source/**``
record tree and the ``plans/**`` frontmatter — into flat, queryable indexes.
 
Responsibilities
----------------
- ``load_source_index`` — id to status/kind/children/prose, plus the changelog
  record bodies, read through the real scanner + record loader.
- ``load_plan_records`` — plan id, path, and the ``advances:`` declaration,
  distinguishing *absent field* from the conformant *declared none*.
- Classify which ids are safe to scan for in prose (``SCANNABLE_ID``).
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - index build lifecycle
    L3 - corpus size summaries
    L4 - per-record classification
 
Contracts
---------
- Read-only: neither loader writes to the corpora it reads.
- The source index is built from ``source/**`` and never from a projection
  store: the on-disk authored projection can be stale, and a check for staleness
  must not take its evidence from a possibly-stale artifact.
- Single-letter and other degenerate roadmap ids are excluded from prose
  scanning and reported in ``unscannable_ids`` — excluded visibly, never
  silently (they collide with agent slot handles and list markers).
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.source.scanner (source tree classification)
- oc_chat.registry.source.canonical (record loading)
- yaml (plan frontmatter parsing)
"""
from __future__ import annotations
 
import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Mapping, Sequence
 
import yaml
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.source.canonical import load_record, split_record_text
from oc_chat.registry.source.scanner import scan_source_tree
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
#: Record kinds that carry a roadmap status.
STATUS_KINDS: frozenset[str] = frozenset({"phase", "promoted_item", "roadmap_item"})
 
#: Statuses that mean "this record will not move again".
TERMINAL_STATUSES: frozenset[str] = frozenset({"done", "deferred", "superseded"})
 
#: Statuses that mean "the landing did move this record".
ADVANCED_STATUSES: frozenset[str] = frozenset({"done", "in_progress"})
 
#: An id is safe to scan for in free prose only if it is at least two characters
#: and carries a digit, dot or hyphen. Bare ``A``/``B``/``C`` are real pre-M0-era
#: item ids AND the workflow's agent slot handles; matching them in prose yields
#: nothing but noise.
SCANNABLE_ID = re.compile(r"^(?=.{2,}$).*[0-9.-]")
 
#: Candidate roadmap-id tokens in prose. Intersected with the real id universe,
#: so this pattern only has to be generous, never precise.
ID_TOKEN = re.compile(r"\b[A-Z][A-Za-z0-9]*(?:[.-][A-Za-z0-9]+)*\b")
 
#: A plan whose ``advances:`` is this token declares, conformantly, that it moves
#: no roadmap record.
ADVANCES_NONE = "none"
 
 
@dataclass(frozen=True)
class RoadmapRecord:
    """One status-bearing roadmap record (a phase, sub-phase, or item)."""
 
    record_id: str
    status: str
    kind: str
    source_ref: str
    prose: str
    parent_id: str | None = None
 
 
@dataclass(frozen=True)
class ChangelogRecord:
    """One ``source/changelog/**`` record: its routing kind and its prose."""
 
    rel_path: str
    kind: str
    date: str
    body: str
 
 
@dataclass(frozen=True)
class SourceIndex:
    """The flat projection of a ``roadmap/source/**`` tree the check reads."""
 
    records: Mapping[str, RoadmapRecord]
    changelog: Mapping[str, ChangelogRecord]
    children: Mapping[str, tuple[str, ...]]
    root: Path
 
    @property
    def scannable_ids(self) -> frozenset[str]:
        """Ids safe to match in free prose."""
 
        return frozenset(
            record_id for record_id in self.records if SCANNABLE_ID.match(record_id)
        )
 
    @property
    def unscannable_ids(self) -> tuple[str, ...]:
        """Ids deliberately excluded from prose scanning, sorted for the payload."""
 
        return tuple(
            sorted(
                record_id
                for record_id in self.records
                if not SCANNABLE_ID.match(record_id)
            )
        )
 
    def status_of(self, record_id: str) -> str | None:
        """The current status of ``record_id``, or None when it has no record."""
 
        record = self.records.get(record_id)
        return None if record is None else record.status
 
 
@dataclass(frozen=True)
class PlanRecord:
    """One plan file's reconciliation-relevant frontmatter."""
 
    plan_id: str
    path: str
    advances: str | None
    has_advances_field: bool
    completed: bool = True
    #: Set when the frontmatter did not parse as strict YAML. The ``advances:``
    #: field is still recovered line-wise, but the failure is reported rather
    #: than silently read as "the plan declared nothing".
    frontmatter_error: str | None = None
 
    @property
    def declares_none(self) -> bool:
        """True when the plan conformantly declares that it moves no record."""
 
        return self.has_advances_field and (
            self.advances is None or self.advances.strip().lower() == ADVANCES_NONE
        )
 
    @property
    def advanced_id(self) -> str | None:
        """The roadmap id this plan declares it advances, if it names one."""
 
        if not self.has_advances_field or self.declares_none:
            return None
        return (self.advances or "").strip()
 
 
def _sub_phase_entries(data: Mapping[str, object]) -> list[tuple[str, str]]:
    """The ``(id, status)`` pairs of a phase record's ``sub_phases`` list."""
 
    entries: list[tuple[str, str]] = []
    for sub in data.get("sub_phases") or []:  # type: ignore[union-attr]
        if isinstance(sub, Mapping) and sub.get("id"):
            entries.append((str(sub["id"]), str(sub.get("status"))))
    return entries
 
 
def load_source_index(source_dir: str | Path) -> SourceIndex:
    """Build the flat index of a ``roadmap/source/**`` tree."""
 
    func = "load_source_index"
    root = Path(source_dir)
    records: dict[str, RoadmapRecord] = {}
    changelog: dict[str, ChangelogRecord] = {}
    children: dict[str, list[str]] = {}
 
    for scanned in scan_source_tree(root):
        model, body = load_record(scanned)
        data = model.model_dump()
        if scanned.kind in STATUS_KINDS:
            record_id = str(data["id"])
            records[record_id] = RoadmapRecord(
                record_id=record_id,
                status=str(data.get("status")),
                kind=scanned.kind,
                source_ref=scanned.rel_path,
                prose=body,
            )
            for sub_id, sub_status in _sub_phase_entries(data):
                records[sub_id] = RoadmapRecord(
                    record_id=sub_id,
                    status=sub_status,
                    kind="sub_phase",
                    source_ref=scanned.rel_path,
                    prose=_sub_phase_prose(data, sub_id),
                    parent_id=record_id,
                )
                children.setdefault(record_id, []).append(sub_id)
        elif scanned.kind == "version":
            changelog[scanned.rel_path] = ChangelogRecord(
                rel_path=scanned.rel_path,
                kind=str(data.get("kind") or ""),
                date=str(data.get("date") or ""),
                body=body,
            )
 
    # A promoted item lives in its own file under ``phases/<PHASE>/`` and is NOT
    # listed in its parent's ``sub_phases``; fold it into the parent's children so
    # class-3 sees the whole sub-part set.
    for record_id, record in records.items():
        if record.kind != "promoted_item" or "." not in record_id:
            continue
        parent = record_id.rsplit(".", 1)[0]
        if parent in records and record_id not in children.get(parent, []):
            children.setdefault(parent, []).append(record_id)
 
    _dbg(
        "OC_CHAT.MAIN",
        func,
        f"indexed {len(records)} records + {len(changelog)} changelog records",
        2,
    )
    return SourceIndex(
        records=records,
        changelog=changelog,
        children={key: tuple(value) for key, value in children.items()},
        root=root,
    )
 
 
def _sub_phase_prose(data: Mapping[str, object], sub_id: str) -> str:
    """The description of ``sub_id`` inside a phase record's ``sub_phases``."""
 
    for sub in data.get("sub_phases") or []:  # type: ignore[union-attr]
        if isinstance(sub, Mapping) and str(sub.get("id")) == sub_id:
            return str(sub.get("description") or "")
    return ""
 
 
#: The ``advances:`` declaration, read line-wise. Plan frontmatter is authored
#: prose-first and is NOT reliably strict YAML (an unquoted ``status:`` line
#: carrying its own ``: `` is enough to break ``safe_load``), so the field is
#: recovered directly rather than via the whole document. Anything else would
#: report "no advances field" for every plan whose *other* keys failed to parse.
_ADVANCES_LINE = re.compile(r"^advances:\s*(.*?)\s*$", re.MULTILINE)
 
#: A plain ``plan_id:`` line, recovered the same way and for the same reason.
_PLAN_ID_LINE = re.compile(r"^plan_id:\s*(.*?)\s*$", re.MULTILINE)
 
 
def _strip_scalar(raw: str) -> str:
    """Strip an inline ``#`` comment and surrounding quotes from a YAML scalar."""
 
    value = raw.split("#", 1)[0].strip() if not raw.startswith(("'", '"')) else raw
    return value.strip().strip("\"'").strip()
 
 
def _plan_frontmatter(text: str) -> tuple[dict[str, object], str, str | None]:
    """Return ``(parsed_mapping, raw_frontmatter, parse_error)`` for a plan file.
 
    The mapping is best-effort: an empty one means the YAML did not parse, NOT
    that the plan declared nothing. Callers must consult ``raw_frontmatter`` for
    the fields they care about and surface ``parse_error`` rather than treating
    a failed parse as an absent field.
    """
 
    try:
        frontmatter, _body = split_record_text(text)
    except Exception as error:  # noqa: BLE001 - a plan without frontmatter is not fatal
        return {}, "", f"no frontmatter: {error}"
    try:
        data = yaml.safe_load(frontmatter) or {}
    except yaml.YAMLError as error:
        first_line = str(error).strip().splitlines()[0]
        return {}, frontmatter, f"frontmatter is not strict YAML: {first_line}"
    return (data if isinstance(data, dict) else {}), frontmatter, None
 
 
def load_plan_records(
    plans_dir: str | Path, *, completed: bool = True
) -> tuple[PlanRecord, ...]:
    """Read every ``*.md`` plan in ``plans_dir`` (non-recursive), sorted by id."""
 
    func = "load_plan_records"
    directory = Path(plans_dir)
    if not directory.is_dir():
        _dbg("OC_CHAT.MAIN", func, f"no plans dir at {directory}", 3)
        return ()
    plans: list[PlanRecord] = []
    for path in sorted(directory.glob("*.md")):
        data, frontmatter, parse_error = _plan_frontmatter(path.read_text(encoding="utf-8"))
        line_match = _ADVANCES_LINE.search(frontmatter)
        if "advances" in data:
            raw = data.get("advances")
            advances = None if raw is None else str(raw)
            has_field = True
        elif line_match is not None:
            advances = _strip_scalar(line_match.group(1)) or None
            has_field = True
        else:
            advances, has_field = None, False
        plan_id = str(data.get("plan_id") or "").strip()
        if not plan_id:
            id_match = _PLAN_ID_LINE.search(frontmatter)
            plan_id = _strip_scalar(id_match.group(1)) if id_match else path.stem
        plans.append(
            PlanRecord(
                plan_id=plan_id or path.stem,
                path=path.name,
                advances=advances,
                has_advances_field=has_field,
                completed=completed,
                frontmatter_error=parse_error,
            )
        )
    _dbg("OC_CHAT.MAIN", func, f"read {len(plans)} plans from {directory.name}", 2)
    return tuple(plans)
 
 
def named_ids(text: str, scannable: frozenset[str]) -> Sequence[str]:
    """The roadmap ids named anywhere in ``text``, sorted."""
 
    return sorted({token for token in ID_TOKEN.findall(text)} & scannable)