plan_metadata.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_orchestrator/plan_metadata.py
 
# OCT-LINT: disable=tests_exist reason="Covered by tests/test_257_dispatcher_seams.py (the provider seam) and oc_chat's own copy of the plan-257 witnesses"
"""
One reader for a plan file's header conventions.
 
Purpose
-------
Plan 257 P2 (A4). A plan file states its branch, worktree, title and
type in a mix of YAML frontmatter and ``**Name:**`` markdown headers,
and the dispatch layer used to parse that in three separate places: the
``**Branch:**`` / ``**Worktree:**`` regexes and the frontmatter path in
``scripts/plan_manage.py``, and the ``# `` / ``**Plan Type:**`` line
scan in ``scripts/plan_status.py``. Three copies of a parser drift.
This module is the one place, and the two call sites now go through it.
 
Responsibilities
----------------
- :func:`split_frontmatter` -- the ``---`` fence split, moved here from
  ``scripts/plan_status.py`` so the format has one splitter in the
  dispatch layer rather than one per consumer.
- :class:`PlanMeta` -- the four header fields plus the parsed
  frontmatter and, when it failed to parse, the error that caused it.
- :class:`PlanMetadataProvider` -- the contract, so a caller can supply
  its own reader.
- :class:`DefaultPlanMetadataProvider` -- today's three parsers,
  consolidated, field for field.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events (a file was read)
    L3 - semantic details (which lane a field came from)
    L4 - deep tracing (module load, per-field parse)
 
Contracts
---------
- The default provider is BYTE-IDENTICAL to the parsers it replaces,
  per field and including their disagreements. ``branch`` and
  ``worktree`` prefer a truthy frontmatter scalar, fall back to an
  unanchored markdown regex, and are returned stripped of whitespace
  and backticks, possibly empty. ``title`` and ``plan_type`` come from
  a whole-file line scan that strips neither backticks nor trailing
  prose, and whose repeat-match precedence depends on ``plan_id``.
  Preserving those quirks is deliberate: plan 257 P6 forbids a
  behaviour change here, and the quirks are reported instead.
- A malformed ``---`` fence yields empty frontmatter, as it always did.
  Invalid YAML is NOT swallowed: it is returned in
  :attr:`PlanMeta.frontmatter_error` so a caller that used to let it
  escape can still let it escape, and a caller that never parsed YAML
  at all can still ignore it.
- Callers own their own fallbacks. This module never synthesises a
  branch, a worktree or a title.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- yaml (frontmatter parse)
"""
 
from __future__ import annotations
 
import re
from collections.abc import Mapping
from dataclasses import dataclass, field
from pathlib import Path
from typing import Protocol
 
import yaml
 
from oc_diagnostics import _dbg
 
_dbg("OC_CHAT.MAIN", "module", "plan_metadata loaded", 4)
 
 
# Markdown-form header parsers (the ``**Branch:**`` / ``**Worktree:**``
# style used by legacy plans without YAML frontmatter). Moved verbatim
# from scripts/plan_manage.py; their known defects are recorded in the
# plan 257 report rather than fixed here.
_BRANCH_RE = re.compile(r"\*\*Branch:\*\*\s*`?(?P<value>[^\s`]+)", re.MULTILINE)
_WORKTREE_RE = re.compile(r"\*\*Worktree:\*\*\s*(?P<value>.+)", re.MULTILINE)
 
_TITLE_PREFIX = "# "
_PLAN_TYPE_PREFIX = "**Plan Type:**"
 
 
@dataclass(frozen=True)
class PlanMeta:
    """A plan file's header fields, as its authors actually wrote them."""
 
    #: Stripped of whitespace and backticks; ``""`` when absent.
    branch: str = ""
    #: Stripped of whitespace and backticks; ``""`` when absent. RAW --
    #: never resolved, because a caller deletes what this names.
    worktree: str = ""
    #: The ``# `` heading text, or ``None`` when there is none.
    title: str | None = None
    #: The ``**Plan Type:**`` value, backticks and trailing prose intact.
    plan_type: str | None = None
    #: The parsed YAML frontmatter; empty when there is none.
    frontmatter: Mapping[str, object] = field(default_factory=dict)
    #: The YAML error, when the frontmatter did not parse.
    frontmatter_error: Exception | None = None
 
 
class PlanMetadataProvider(Protocol):
    """Reads a plan file's header fields."""
 
    def metadata(self, plan_path: Path, *, plan_id: str | None = None) -> PlanMeta:
        """Read *plan_path* and return its header fields."""
 
    def metadata_from_text(self, text: str, *, plan_id: str | None = None) -> PlanMeta:
        """Return the header fields of an already-read plan file."""
 
 
def split_frontmatter(text: str) -> tuple[str, str]:
    """Split text into ``(yaml_frontmatter, markdown_body)``.
 
    Moved from ``scripts/plan_status.py`` (plan 257 P2), which still
    exposes it under its old private name for the two callers that
    reach it that way.
    """
    func = "split_frontmatter"
    lines = text.splitlines(keepends=True)
    if not lines or lines[0].rstrip("\r\n") != "---":
        raise ValueError("missing opening '---' delimiter")
    end = None
    for i in range(1, len(lines)):
        if lines[i].rstrip("\r\n") == "---":
            end = i
            break
    if end is None:
        raise ValueError("missing closing '---' delimiter")
    frontmatter = "".join(lines[1:end])
    body = "".join(lines[end + 1 :]).lstrip("\n")
    _dbg("OC_CHAT.MAIN", func, f"frontmatter {len(frontmatter)}B body {len(body)}B", 4)
    return frontmatter, body
 
 
def _markdown_value(text: str, pattern: re.Pattern[str]) -> str | None:
    """First match of *pattern*, stripped of whitespace and backticks."""
    match = pattern.search(text)
    if not match:
        return None
    return match.group("value").strip().strip("`").strip()
 
 
def _scan_headings(text: str, plan_id: str | None) -> tuple[str | None, str | None]:
    """The ``# `` title and ``**Plan Type:**`` value, in one pass.
 
    One pass, with one shared stop condition, because that is what the
    pre-257 loop did and the stop condition is observable: it compares
    the title against *plan_id*, so repeat-match precedence differs
    depending on whether the heading happens to equal the plan id.
    """
    func = "_scan_headings"
    title: str | None = None
    plan_type: str | None = None
    for line in text.splitlines():
        stripped = line.strip()
        if stripped.startswith(_TITLE_PREFIX):
            title = stripped[len(_TITLE_PREFIX) :].strip() or None
        if stripped.startswith(_PLAN_TYPE_PREFIX):
            plan_type = stripped.split(_PLAN_TYPE_PREFIX, 1)[1].strip() or None
        if title is not None and title != plan_id and plan_type is not None:
            break
    _dbg("OC_CHAT.MAIN", func, f"title={title!r} plan_type={plan_type!r}", 4)
    return title, plan_type
 
 
class DefaultPlanMetadataProvider:
    """The three pre-257 parsers, consolidated without changing a verdict."""
 
    def metadata(self, plan_path: Path, *, plan_id: str | None = None) -> PlanMeta:
        """Read *plan_path* as UTF-8 and parse its header."""
        func = "DefaultPlanMetadataProvider.metadata"
        text = Path(plan_path).read_text(encoding="utf-8")
        _dbg("OC_CHAT.MAIN", func, f"read {plan_path} ({len(text)}B)", 2)
        return self.metadata_from_text(text, plan_id=plan_id)
 
    def metadata_from_text(self, text: str, *, plan_id: str | None = None) -> PlanMeta:
        """Parse *text*; reads nothing from disk."""
        frontmatter, error = _parse_frontmatter(text)
        branch = str(frontmatter.get("branch") or _markdown_value(text, _BRANCH_RE) or "").strip()
        worktree = str(
            frontmatter.get("worktree") or _markdown_value(text, _WORKTREE_RE) or ""
        ).strip()
        title, plan_type = _scan_headings(text, plan_id)
        return PlanMeta(
            branch=branch,
            worktree=worktree,
            title=title,
            plan_type=plan_type,
            frontmatter=frontmatter,
            frontmatter_error=error,
        )
 
 
def _parse_frontmatter(text: str) -> tuple[dict[str, object], Exception | None]:
    """The frontmatter mapping, and the YAML error that stopped it.
 
    The ``ValueError`` arm is the pre-257 "no frontmatter -- markdown-form
    plan" path. The ``YAMLError`` arm is new only in that the error is
    returned rather than raised; ``plan_manage`` re-raises it, so the
    caller that used to see it still does, and ``plan_status``, which
    never parsed YAML here at all, is unaffected by a plan whose prose
    frontmatter is not strict YAML.
    """
    func = "_parse_frontmatter"
    try:
        raw, _body = split_frontmatter(text)
        loaded = yaml.safe_load(raw)
    except ValueError:
        _dbg("OC_CHAT.MAIN", func, "no frontmatter fence; markdown-form plan", 3)
        return {}, None
    except yaml.YAMLError as exc:
        # VALIDATION_ERROR: a hand-authored plan header that is not strict
        # YAML. Returned, not raised, so each caller keeps its own policy.
        _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: frontmatter: {exc}", 3)
        return {}, exc
    if isinstance(loaded, dict):
        return loaded, None
    return {}, None
 
 
#: The provider every call site uses unless it is handed another one.
DEFAULT_PROVIDER: PlanMetadataProvider = DefaultPlanMetadataProvider()
 
 
def resolve_provider(provider: PlanMetadataProvider | None) -> PlanMetadataProvider:
    """*provider*, or the module default when it is ``None``."""
    return provider if provider is not None else DEFAULT_PROVIDER
 
 
__all__ = [
    "DEFAULT_PROVIDER",
    "DefaultPlanMetadataProvider",
    "PlanMeta",
    "PlanMetadataProvider",
    "resolve_provider",
    "split_frontmatter",
]