canonical.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/source/canonical.py
 
# OCT-LINT: disable=no_hardcoded_secrets reason="YAML tag URIs and field names are format constants"
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/source/test_canonical.py"
"""
Canonical YAML dumper + record loader for the AuthoredSource format.
 
Purpose
-------
Read a source record (``yaml.safe_load``) and re-emit byte-canonical frontmatter
with a custom ``yaml.SafeDumper`` subclass, per Registry Source & Storage Spec
V0.22 §3.2: scalar string values double-quoted, plain keys, field-declaration
key order, pydantic defaults + empty containers omitted, ``depends_on``
canonical emission, no anchors/aliases/tags, LF / UTF-8 / no-BOM / one trailing
newline. The formatter is idempotent (AC §8.4).
 
Responsibilities
----------------
- ``format_record_text(model, body)`` — canonical bytes for a record.
- ``load_record(scanned)`` — parse + strict-validate against the P1.1 model.
- ``canonicalize_file`` / ``format_source_tree`` — format or ``--check`` a tree.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - format/load lifecycle
    L3 - per-record canonicalisation
    L4 - YAML node emission details
 
Contracts
---------
- Idempotent: ``format(format(x)) == format(x)``.
- Round-trip safe: a canonical record re-parses to an equal model.
- All string scalar VALUES double-quoted (round-trips any single-line content);
  keys emitted plain. ``depends_on`` blocks → bare quoted id; enables/gates →
  inline ``{id: "..", kind: ..}`` (never ``{id, kind: blocks}``).
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- yaml (safe_load + custom SafeDumper)
- option_c_registry_contracts (P1.1 entity models)
- oc_chat.registry.source.errors (SRC_* codes)
- oc_chat.registry.source.promoted_item (PromotedItem - the model for promoted_item records)
- oc_chat.registry.source.scanner (ScannedRecord - the scanned record every loader and formatter takes)
- pydantic (ValidationError - reported as an SRC_STRICT_TYPE failure)
- pydantic_core (PydanticUndefined - telling a field with no default apart)
"""
from __future__ import annotations
 
from typing import Any, Optional
 
import yaml
from pydantic import ValidationError
from pydantic_core import PydanticUndefined
 
from oc_diagnostics import _dbg
 
from option_c_registry_contracts import (
    DependencyEdge,
    RegistryModel,
    SOURCE_ENTITIES,
)
 
from oc_chat.registry.source.errors import (
    SRC_BAD_YAML,
    SRC_BAD_YAML_FEATURE,
    SRC_DUP_OR_STEM_MISMATCH,
    SRC_NOT_CANONICAL,
    SRC_STRICT_TYPE,
    SourceValidationError,
)
from oc_chat.registry.source.promoted_item import PromotedItem
from oc_chat.registry.source.scanner import ScannedRecord
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
_STR_TAG = "tag:yaml.org,2002:str"
 
#: Reserved Windows device stems (§3.2) — a record whose stem's first dot-segment
#: is one of these cannot be checked out on Windows. Rejected per-record on load
#: (Plan 142j R7 / VC-0007, layer 2). Reuses SRC_DUP_OR_STEM_MISMATCH (a stem
#: validity failure; SRC_UNSAFE_PATH is a TARGET code not in ALL_CODES).
_RESERVED_DEVICE_STEMS: frozenset[str] = frozenset(
    {"con", "prn", "aux", "nul"}
    | {f"com{i}" for i in range(1, 10)}
    | {f"lpt{i}" for i in range(1, 10)}
)
 
 
class _FlowMap(dict):
    """A mapping rendered inline (flow style) — used for ``depends_on`` edges."""
 
 
class CanonicalDumper(yaml.SafeDumper):
    """SafeDumper that double-quotes string values and emits plain keys."""
 
    def represent_mapping(
        self,
        tag: str,
        mapping: dict[str, Any],
        flow_style: Optional[bool] = None,
    ) -> yaml.MappingNode:
        """Emit a mapping with plain keys + (recursively) quoted scalar values."""
        value = []
        node = yaml.MappingNode(tag, value, flow_style=flow_style)
        if self.alias_key is not None:
            self.represented_objects[self.alias_key] = node
        for item_key, item_value in mapping.items():
            node_key = self.represent_scalar(_STR_TAG, str(item_key), style="")
            node_value = self.represent_data(item_value)
            value.append((node_key, node_value))
        if flow_style is None:
            node.flow_style = False
        return node
 
 
def _represent_str(dumper: yaml.SafeDumper, data: str) -> yaml.ScalarNode:
    # §3.2: a multi-line string value (the [NEW·V24] era.banner / framing.text
    # verbatim prose) is emitted as a literal block scalar (``|``); a single-line
    # value stays double-quoted (round-trips any single-line content). PyYAML
    # falls back to a quoted style automatically if a block scalar cannot
    # represent the content (e.g. trailing whitespace) — the canonicaliser
    # normalises those away, so block scalars stay clean + idempotent.
    if "\n" in data:
        return dumper.represent_scalar(_STR_TAG, data, style="|")
    return dumper.represent_scalar(_STR_TAG, data, style='"')
 
 
def _represent_flowmap(dumper: yaml.SafeDumper, data: _FlowMap) -> yaml.MappingNode:
    return dumper.represent_mapping("tag:yaml.org,2002:map", data, flow_style=True)
 
 
CanonicalDumper.add_representer(str, _represent_str)
CanonicalDumper.add_representer(_FlowMap, _represent_flowmap)
 
 
def _field_default(field_info: Any) -> Any:
    if field_info.default_factory is not None:  # type: ignore[truthy-function]
        return field_info.default_factory()
    if field_info.default is not PydanticUndefined:
        return field_info.default
    return PydanticUndefined
 
 
def _to_yaml_obj(value: Any) -> Any:
    if isinstance(value, RegistryModel):
        return model_to_frontmatter(value)
    if isinstance(value, list):
        return [_to_yaml_obj(item) for item in value]
    if isinstance(value, dict):
        return {key: _to_yaml_obj(item) for key, item in value.items()}
    return value
 
 
def _depends_on_to_yaml(edge: Any) -> Any:
    """Canonical ``depends_on`` emission (§3.2 / AC §8.36)."""
    if isinstance(edge, DependencyEdge):
        if edge.kind == "blocks":
            return edge.id  # bare quoted id — never {id, kind: blocks}
        return _FlowMap({"id": edge.id, "kind": edge.kind})
    return edge  # already a bare id str (kind defaults to blocks)
 
 
def model_to_frontmatter(model: RegistryModel) -> dict[str, Any]:
    """Build the frontmatter mapping in field-declaration order, omitting
    pydantic defaults and empty containers (§3.2)."""
    out: dict[str, Any] = {}
    for name, field_info in type(model).model_fields.items():
        value = getattr(model, name)
        if value is None:
            continue
        if isinstance(value, (list, dict)) and not value:
            continue
        default = _field_default(field_info)
        if (
            default is not PydanticUndefined
            and not isinstance(value, (list, dict))
            and value == default
        ):
            continue
        if name == "depends_on":
            out[name] = [_depends_on_to_yaml(edge) for edge in value]
        else:
            out[name] = _to_yaml_obj(value)
    return out
 
 
def dump_frontmatter(mapping: dict[str, Any]) -> str:
    """Serialize a frontmatter mapping to canonical YAML text (LF, no BOM)."""
    return yaml.dump(
        mapping,
        Dumper=CanonicalDumper,
        default_flow_style=False,
        sort_keys=False,
        allow_unicode=True,
        width=4096,
    )
 
 
def _normalize_body(body: str) -> str:
    body = body.replace("\r\n", "\n").replace("\r", "\n")
    body = body.strip("\n")
    return "\n".join(line.rstrip() for line in body.split("\n")).strip("\n")
 
 
def format_record_text(model: RegistryModel, body: str = "") -> str:
    """Return the canonical bytes (text) of a record: frontmatter + body."""
    frontmatter = dump_frontmatter(model_to_frontmatter(model))
    clean_body = _normalize_body(body)
    if clean_body:
        return f"---\n{frontmatter}---\n\n{clean_body}\n"
    return f"---\n{frontmatter}---\n"
 
 
def split_record_text(text: str) -> tuple[str, str]:
    """Split a record file into (frontmatter_text, body); LF-normalised."""
    text = text.replace("\r\n", "\n").replace("\r", "\n")
    lines = text.split("\n")
    if not lines or lines[0].strip() != "---":
        raise SourceValidationError(
            "record has no '---' YAML frontmatter delimiter",
            code=SRC_BAD_YAML,
            hint="every source record starts with a '---' frontmatter block",
        )
    for index in range(1, len(lines)):
        if lines[index].strip() == "---":
            frontmatter = "\n".join(lines[1:index]) + "\n"
            body = "\n".join(lines[index + 1 :])
            return frontmatter, body
    raise SourceValidationError(
        "unterminated YAML frontmatter (missing closing '---')",
        code=SRC_BAD_YAML,
    )
 
 
def _assert_no_yaml_features(frontmatter: str, rel_path: str) -> None:
    """Reject anchors/aliases/merge/custom-tags (§3.2) → SRC_BAD_YAML_FEATURE."""
    try:
        events = list(yaml.parse(frontmatter, Loader=yaml.SafeLoader))
    except yaml.YAMLError as error:
        raise SourceValidationError(
            f"{rel_path}: unparseable YAML frontmatter: {error}",
            code=SRC_BAD_YAML,
            records=[rel_path],
        ) from error
    for event in events:
        if isinstance(event, yaml.AliasEvent):
            raise SourceValidationError(
                f"{rel_path}: YAML aliases/merge keys are forbidden in frontmatter",
                code=SRC_BAD_YAML_FEATURE,
                records=[rel_path],
            )
        if getattr(event, "anchor", None):
            raise SourceValidationError(
                f"{rel_path}: YAML anchors are forbidden in frontmatter",
                code=SRC_BAD_YAML_FEATURE,
                records=[rel_path],
            )
        tag = getattr(event, "tag", None)
        if tag and isinstance(tag, str) and tag.startswith("!"):
            raise SourceValidationError(
                f"{rel_path}: explicit YAML tags are forbidden in frontmatter",
                code=SRC_BAD_YAML_FEATURE,
                records=[rel_path],
            )
 
 
def _model_for(scanned: ScannedRecord) -> Any:
    if scanned.kind in SOURCE_ENTITIES:
        return SOURCE_ENTITIES[scanned.kind]
    if scanned.kind == "promoted_item":
        # [OP-5] promoted parity: resolve the relocated PromotedItem model so
        # promoted records canonicalise / format / hash like every other entity.
        # (Not in SOURCE_ENTITIES — that closed dict lives in the contracts pkg.)
        return PromotedItem
    raise SourceValidationError(
        f"{scanned.rel_path}: no source model for kind {scanned.kind!r}",
        code=SRC_STRICT_TYPE,
        records=[scanned.rel_path],
    )
 
 
def load_record(scanned: ScannedRecord) -> tuple[RegistryModel, str]:
    """Load + strict-validate a scanned record into its P1.1 model + body."""
    try:
        raw = scanned.path.read_text(encoding="utf-8")
    except UnicodeDecodeError as error:
        raise SourceValidationError(
            f"{scanned.rel_path}: not valid UTF-8: {error}",
            code=SRC_BAD_YAML,
            records=[scanned.rel_path],
        ) from error
    frontmatter, body = split_record_text(raw)
    _assert_no_yaml_features(frontmatter, scanned.rel_path)
    try:
        data = yaml.safe_load(frontmatter) or {}
    except yaml.YAMLError as error:
        raise SourceValidationError(
            f"{scanned.rel_path}: unparseable YAML frontmatter: {error}",
            code=SRC_BAD_YAML,
            records=[scanned.rel_path],
        ) from error
    if not isinstance(data, dict):
        raise SourceValidationError(
            f"{scanned.rel_path}: frontmatter must be a mapping",
            code=SRC_BAD_YAML,
            records=[scanned.rel_path],
        )
    model_cls = _model_for(scanned)
    try:
        model = model_cls.model_validate(data)
    except ValidationError as error:
        raise SourceValidationError(
            f"{scanned.rel_path}: strict validation failed: {error}",
            code=SRC_STRICT_TYPE,
            records=[scanned.rel_path],
        ) from error
    _check_identity(scanned, model)
    return model, _normalize_body(body)
 
 
def _check_identity(scanned: ScannedRecord, model: RegistryModel) -> None:
    """Filename stem == frontmatter id (§4); items/<kind>/ matches track_kind.
 
    [NEW·V24] A changelog ``version`` record's identity is its ``slug`` field
    (§3.5 / §4 item 6), so its stem is now checked against ``slug`` (the earlier
    ``version``-skip is removed); ``framing`` records check stem == ``id`` via the
    generic path. Only ``manifest`` (which has no id) is exempt.
    """
    if scanned.stem.split(".", 1)[0].casefold() in _RESERVED_DEVICE_STEMS:
        # Plan 142j R7 (VC-0007, layer 2): per-record reserved Windows device stem.
        raise SourceValidationError(
            f"{scanned.rel_path}: reserved Windows device stem {scanned.stem!r} "
            "cannot be checked out on Windows",
            code=SRC_DUP_OR_STEM_MISMATCH,
            records=[scanned.rel_path],
            hint="rename the record — CON/PRN/AUX/NUL/COM1-9/LPT1-9 are reserved (§3.2)",
        )
    record_id = (
        getattr(model, "id", None)
        or getattr(model, "track_code", None)
        or getattr(model, "slug", None)
    )
    if scanned.kind != "manifest" and record_id is not None:
        if record_id != scanned.stem:
            raise SourceValidationError(
                f"{scanned.rel_path}: filename stem {scanned.stem!r} != id {record_id!r}",
                code=SRC_DUP_OR_STEM_MISMATCH,
                records=[scanned.rel_path],
            )
    if scanned.kind == "roadmap_item":
        track_kind = getattr(model, "track_kind", None)
        if track_kind != scanned.track_kind:
            raise SourceValidationError(
                f"{scanned.rel_path}: track_kind {track_kind!r} != directory "
                f"items/{scanned.track_kind}/",
                code=SRC_STRICT_TYPE,
                records=[scanned.rel_path],
            )
 
 
def canonicalize_file(scanned: ScannedRecord) -> str:
    """Return the canonical text for a scanned record (load + re-emit)."""
    model, body = load_record(scanned)
    return format_record_text(model, body)
 
 
def format_source_tree(
    records: list[ScannedRecord], *, check: bool = False
) -> list[str]:
    """Format (or, with ``check=True``, verify) every formattable record.
 
    Returns the list of relative paths that were rewritten (or, in check mode,
    the non-canonical paths). [OP-5] promoted parity: promoted items are
    canonicalised + ``--check``-verified like every other entity (the earlier
    model-gap skip is removed) so ``format --check`` catches a non-canonical
    promoted file.
    """
    changed: list[str] = []
    for scanned in records:
        canonical = canonicalize_file(scanned)
        current = scanned.path.read_text(encoding="utf-8")
        if current == canonical:
            continue
        changed.append(scanned.rel_path)
        if not check:
            scanned.path.write_text(canonical, encoding="utf-8", newline="")
    if check and changed:
        raise SourceValidationError(
            f"{len(changed)} record(s) are not canonical: {changed}",
            code=SRC_NOT_CANONICAL,
            records=changed,
            hint="run 'oct agent roadmap source format' to canonicalise",
        )
    return changed
 
 
def canonical_record_bytes(scanned: ScannedRecord) -> bytes:
    """The canonical UTF-8 bytes for a record (source-tree-hash preimage input)."""
    return canonicalize_file(scanned).encode("utf-8")