history.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/roadmap/reconcile/history.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/roadmap/test_reconcile_history.py"
"""
Git-history reader for the roadmap reconciliation check (Plan 205).
 
Purpose
-------
Recover, per commit, the two facts the check joins on: which ``source/changelog``
records the commit ADDED, and which roadmap record statuses the same commit MOVED.
 
Responsibilities
----------------
- ``read_history`` — one ``git log`` pass for added changelog records, then a
  status-map diff only for the commits that also touched ``phases``/``items``.
- Recover nested ``sub_phases`` transitions **by id**, by loading the whole
  record before and after the commit rather than parsing a unified diff (a
  ``-U0`` hunk cannot reliably associate a changed ``status:`` with its ``id:``).
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - history read lifecycle
    L3 - per-commit summaries
    L4 - per-blob status-map details
 
Contracts
---------
- Read-only: every git invocation is a query (``log`` / ``show``); the reader
  never writes to the repository, the index, or the working tree.
- All git output is decoded as UTF-8 with replacement — roadmap records carry
  em-dashes and emoji, and the Windows console codepage must never corrupt them.
- A record that cannot be parsed at some historical revision contributes no
  status entries rather than raising: history predates the current schema.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.source.canonical (frontmatter splitting)
- yaml (historical frontmatter parsing)
"""
from __future__ import annotations
 
import subprocess
from dataclasses import dataclass, field
from pathlib import Path
from typing import Mapping
 
import yaml
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.source.canonical import split_record_text
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
#: Source-tree-relative subdirectories that carry status-bearing records.
STATUS_DIRS: tuple[str, ...] = ("phases", "items")
 
#: Source-tree-relative subdirectory that carries changelog records.
CHANGELOG_DIR = "changelog"
 
_RECORD_SEP = "\x01"
_FIELD_SEP = "\x1f"
 
 
class HistoryError(RuntimeError):
    """Raised when the repository cannot be read (not when a record cannot parse)."""
 
 
@dataclass(frozen=True)
class StatusChange:
    """One roadmap record status movement inside a single commit."""
 
    record_id: str
    from_status: str | None
    to_status: str
    path: str
 
 
@dataclass(frozen=True)
class CommitRecord:
    """One commit, projected to the two facts the reconciliation check joins on.
 
    ``bodies`` carries each added record's prose **as it was written in this
    commit**. The claim that matters is the one the round actually made; reading
    today's text instead would judge a past commit by a later edit. It is
    optional so the engine stays injectable without git.
    """
 
    sha: str
    date: str
    subject: str
    added_changelog: tuple[str, ...]
    status_changes: tuple[StatusChange, ...]
    bodies: Mapping[str, str] = field(default_factory=dict)
 
    @property
    def moved_ids(self) -> frozenset[str]:
        """The set of roadmap record ids whose status moved in this commit."""
 
        return frozenset(change.record_id for change in self.status_changes)
 
 
def _git(repo_root: Path, *args: str) -> str:
    """Run a read-only git query in ``repo_root`` and return its UTF-8 stdout."""
 
    try:
        completed = subprocess.run(
            ["git", *args],
            cwd=str(repo_root),
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            check=True,
        )
    except FileNotFoundError as error:  # git absent from PATH
        raise HistoryError("git executable not found on PATH") from error
    except subprocess.CalledProcessError as error:
        raise HistoryError(
            f"git {' '.join(args[:2])} failed in {repo_root}: {error.stderr.strip()}"
        ) from error
    return completed.stdout
 
 
def _blob(repo_root: Path, revision: str, path: str) -> str | None:
    """Return the blob at ``revision:path``, or None when it does not exist there."""
 
    completed = subprocess.run(
        ["git", "show", f"{revision}:{path}"],
        cwd=str(repo_root),
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        check=False,
    )
    if completed.returncode != 0:
        return None
    return completed.stdout
 
 
def status_map(text: str | None) -> dict[str, str]:
    """Project one record's text to ``{record_id: status}`` (top level + sub-phases).
 
    Returns an empty map for absent or unparseable text: a record that predates
    the current schema contributes no evidence rather than failing the run.
    """
 
    if not text or not text.strip():
        return {}
    try:
        frontmatter, _body = split_record_text(text)
        data = yaml.safe_load(frontmatter) or {}
    except Exception:  # noqa: BLE001 - historical revisions may not parse
        return {}
    if not isinstance(data, dict) or not data.get("id"):
        return {}
    statuses = {str(data["id"]): str(data.get("status"))}
    for sub in data.get("sub_phases") or []:
        if isinstance(sub, dict) and sub.get("id"):
            statuses[str(sub["id"])] = str(sub.get("status"))
    return statuses
 
 
def _changed_status_paths(repo_root: Path, sha: str, source_rel: str) -> list[str]:
    """Paths under ``phases``/``items`` that ``sha`` touched."""
 
    pathspecs = [f"{source_rel}/{name}" for name in STATUS_DIRS]
    output = _git(repo_root, "show", "--format=", "--name-only", sha, "--", *pathspecs)
    return [line.strip() for line in output.splitlines() if line.strip()]
 
 
def _status_changes(repo_root: Path, sha: str, source_rel: str) -> tuple[StatusChange, ...]:
    """Diff the before/after status maps of every status-bearing path in ``sha``."""
 
    changes: list[StatusChange] = []
    for path in _changed_status_paths(repo_root, sha, source_rel):
        before = status_map(_blob(repo_root, f"{sha}^", path))
        after = status_map(_blob(repo_root, sha, path))
        for record_id, to_status in sorted(after.items()):
            if before.get(record_id) == to_status:
                continue
            changes.append(
                StatusChange(
                    record_id=record_id,
                    from_status=before.get(record_id),
                    to_status=to_status,
                    path=path,
                )
            )
    return tuple(changes)
 
 
def _added_bodies(
    repo_root: Path, sha: str, prefix: str, added: tuple[str, ...]
) -> dict[str, str]:
    """The prose of each record ``sha`` added, as written in that commit."""
 
    bodies: dict[str, str] = {}
    for rel in added:
        text = _blob(repo_root, sha, prefix + rel)
        if not text:
            continue
        try:
            _frontmatter, body = split_record_text(text)
        except Exception:  # noqa: BLE001 - a historical record may not split
            body = text
        bodies[rel] = body
    return bodies
 
 
def read_history(
    repo_root: str | Path,
    *,
    source_rel: str = "oc_chat/oc_chat/roadmap/source",
    since: str | None = None,
) -> tuple[CommitRecord, ...]:
    """Read the commits that added changelog records, newest first.
 
    ``since`` is passed straight to ``git log --since`` (any git date spec).
    Only commits that added at least one ``changelog/**`` record are returned:
    those are the commits that make a claim, and the check has nothing to say
    about the others.
    """
 
    func = "read_history"
    root = Path(repo_root)
    args = [
        "log",
        "--diff-filter=A",
        "--name-only",
        f"--format={_RECORD_SEP}%H{_FIELD_SEP}%ad{_FIELD_SEP}%s",
        "--date=short",
    ]
    if since:
        args.append(f"--since={since}")
    args += ["--", f"{source_rel}/{CHANGELOG_DIR}"]
    output = _git(root, *args)
 
    commits: list[CommitRecord] = []
    prefix = f"{source_rel}/"
    for block in output.split(_RECORD_SEP):
        if not block.strip():
            continue
        head, _, rest = block.partition("\n")
        parts = head.split(_FIELD_SEP)
        if len(parts) != 3:
            continue
        sha, date, subject = parts
        added = tuple(
            line.strip()[len(prefix):]
            for line in rest.splitlines()
            if line.strip().startswith(prefix)
        )
        if not added:
            continue
        commits.append(
            CommitRecord(
                sha=sha,
                date=date,
                subject=subject,
                added_changelog=added,
                status_changes=_status_changes(root, sha, source_rel),
                bodies=_added_bodies(root, sha, prefix, added),
            )
        )
    _dbg("OC_CHAT.MAIN", func, f"read {len(commits)} claim-bearing commits", 2)
    return tuple(commits)