core/formation/compose.py

"""Phase 4 — Compose: ValidatedTripleSet -> CourseYAML, byte-stable.

The composer is deterministic by construction:

* The template renders a JSON-shaped dict whose contents are independent of
  input ordering (concepts lex-sorted; relations topo-sorted with lex
  tie-break; sources lex-sorted within each candidate).
* The YAML emitter walks the dict in sorted-key order and uses a fixed
  block style with explicit list markers.  Strings, ints, and bools only —
  floats are forbidden (per ``formation.hashing``).
* ``course_sha256 = sha256(yaml_bytes)``.

If PyYAML is importable we additionally cross-check that
``yaml.safe_load(yaml_bytes)`` round-trips to the same dict; otherwise the
emitter is the sole source of truth.  Round-trip parity is enforced under
test as well.
"""

from __future__ import annotations

import hashlib
import re
from typing import Any

from formation.course import CourseYAML, SubjectSpec, ValidatedTripleSet
from formation.hashing import _reject_floats, sha256_of
from formation.templates import get_template

_PLAIN_SAFE = re.compile(r"^[A-Za-z_][A-Za-z0-9_./:-]*$")
_NEEDS_QUOTING_KEYWORDS = frozenset(
    {
        "",
        "true",
        "false",
        "yes",
        "no",
        "on",
        "off",
        "null",
        "~",
        "True",
        "False",
        "Yes",
        "No",
        "On",
        "Off",
        "Null",
        "NULL",
    }
)


def compose(
    validated_set: ValidatedTripleSet,
    spec: SubjectSpec,
    source_bundle_sha: str,
    template_id: str = "definition",
    template_version: str = "1.0.0",
) -> CourseYAML:
    """Compose a deterministic :class:`CourseYAML` artifact.

    Same logical input -> same ``yaml_bytes`` -> same ``course_sha256``.
    Reordering the relations/concepts/counters tuples does not change the
    output because the template re-sorts them.

    Args:
        validated_set: Forge output.  All triples have passed every rule.
        spec: Subject spec (course header info).
        source_bundle_sha: SHA of the ore bundle that fed the Forge.
        template_id: Which template to render with (default ``"definition"``).
        template_version: Caller-asserted template version; must match the
            template's own ``template_version``.  This propagates into the
            YAML body and therefore into ``course_sha256``.
    """
    template = get_template(template_id)
    if template.template_version != template_version:
        raise ValueError(
            f"compose: template_version mismatch — caller asked for "
            f"{template_version!r} but registry has "
            f"{template.template_version!r}"
        )

    body = template.render(
        validated_set=validated_set,
        spec=spec,
        source_bundle_sha=source_bundle_sha,
    )
    _reject_floats(body)  # belt and suspenders — same rule as canonical_json
    yaml_bytes = _emit_canonical_yaml(body)
    course_sha256 = hashlib.sha256(yaml_bytes).hexdigest()
    course_id = str(body["course_id"])
    return CourseYAML(
        course_id=course_id,
        yaml_bytes=yaml_bytes,
        course_sha256=course_sha256,
        source_bundle_sha=source_bundle_sha,
        validated_set_sha=sha256_of(_validated_set_payload(validated_set)),
        template_id=template.template_id,
        template_version=template.template_version,
    )


# ---------- canonical YAML emitter ----------


def _emit_canonical_yaml(payload: dict[str, Any]) -> bytes:
    """Emit ``payload`` as deterministic block-style YAML bytes.

    Supports strings, ints, bools, lists, and dicts only.  Keys are emitted
    in sorted order; floats raise (rejected upstream too).  No anchors, no
    tags, no flow style — only block scalars and block sequences/mappings.
    """
    lines: list[str] = []
    _emit_node(payload, indent=0, lines=lines, in_list=False)
    return ("\n".join(lines) + "\n").encode("utf-8")


def _emit_node(
    node: Any, *, indent: int, lines: list[str], in_list: bool
) -> None:
    pad = "  " * indent
    if isinstance(node, dict):
        if not node:
            lines.append(f"{pad}{{}}")
            return
        for key in sorted(node.keys()):
            if not isinstance(key, str):
                raise TypeError(
                    "canonical YAML: dict keys must be strings, got "
                    f"{type(key).__name__}"
                )
            value = node[key]
            key_repr = _scalar(key, force_quote_keyword=True)
            if isinstance(value, dict):
                if not value:
                    lines.append(f"{pad}{key_repr}: {{}}")
                else:
                    lines.append(f"{pad}{key_repr}:")
                    _emit_node(value, indent=indent + 1, lines=lines, in_list=False)
            elif isinstance(value, list):
                if not value:
                    lines.append(f"{pad}{key_repr}: []")
                else:
                    lines.append(f"{pad}{key_repr}:")
                    _emit_list(value, indent=indent, lines=lines)
            else:
                lines.append(f"{pad}{key_repr}: {_scalar(value)}")
    elif isinstance(node, list):
        if not node:
            lines.append(f"{pad}[]")
            return
        _emit_list(node, indent=indent, lines=lines)
    else:
        lines.append(f"{pad}{_scalar(node)}")


def _emit_list(items: list[Any], *, indent: int, lines: list[str]) -> None:
    pad = "  " * indent
    for item in items:
        if isinstance(item, dict):
            if not item:
                lines.append(f"{pad}- {{}}")
                continue
            keys = sorted(item.keys())
            first_key = keys[0]
            first_value = item[first_key]
            first_repr = _scalar(first_key, force_quote_keyword=True)
            if isinstance(first_value, dict):
                if not first_value:
                    lines.append(f"{pad}- {first_repr}: {{}}")
                else:
                    lines.append(f"{pad}- {first_repr}:")
                    _emit_node(
                        first_value, indent=indent + 2, lines=lines, in_list=False
                    )
            elif isinstance(first_value, list):
                if not first_value:
                    lines.append(f"{pad}- {first_repr}: []")
                else:
                    lines.append(f"{pad}- {first_repr}:")
                    _emit_list(first_value, indent=indent + 2, lines=lines)
            else:
                lines.append(f"{pad}- {first_repr}: {_scalar(first_value)}")
            for k in keys[1:]:
                v = item[k]
                k_repr = _scalar(k, force_quote_keyword=True)
                inner_pad = "  " * (indent + 1)
                if isinstance(v, dict):
                    if not v:
                        lines.append(f"{inner_pad}{k_repr}: {{}}")
                    else:
                        lines.append(f"{inner_pad}{k_repr}:")
                        _emit_node(v, indent=indent + 2, lines=lines, in_list=False)
                elif isinstance(v, list):
                    if not v:
                        lines.append(f"{inner_pad}{k_repr}: []")
                    else:
                        lines.append(f"{inner_pad}{k_repr}:")
                        _emit_list(v, indent=indent + 2, lines=lines)
                else:
                    lines.append(f"{inner_pad}{k_repr}: {_scalar(v)}")
        elif isinstance(item, list):
            if not item:
                lines.append(f"{pad}- []")
            else:
                lines.append(f"{pad}-")
                _emit_list(item, indent=indent + 1, lines=lines)
        else:
            lines.append(f"{pad}- {_scalar(item)}")


def _scalar(value: Any, *, force_quote_keyword: bool = False) -> str:
    """Emit a scalar.  Strings are quoted unless plain-safe and unambiguous."""
    if isinstance(value, bool):
        # bool is a subclass of int — check first.
        return "true" if value else "false"
    if isinstance(value, int):
        return str(value)
    if value is None:
        return '"null"'  # we forbid bare null to keep parser-agnostic
    if isinstance(value, float):
        raise TypeError(
            "canonical YAML: float values are forbidden; encode as strings"
        )
    if not isinstance(value, str):
        raise TypeError(
            f"canonical YAML: unsupported scalar type {type(value).__name__}"
        )
    needs_quote = (
        force_quote_keyword
        and value in _NEEDS_QUOTING_KEYWORDS
    )
    if needs_quote or _needs_quoting(value):
        return _double_quoted(value)
    return value


def _needs_quoting(text: str) -> bool:
    if text == "":
        return True
    if text in _NEEDS_QUOTING_KEYWORDS:
        return True
    if _PLAIN_SAFE.match(text) is None:
        return True
    # Avoid being parsed as an int by YAML readers.
    if text.lstrip("-").isdigit():
        return True
    return False


def _double_quoted(text: str) -> str:
    escaped = (
        text.replace("\\", "\\\\")
        .replace('"', '\\"')
        .replace("\n", "\\n")
        .replace("\t", "\\t")
        .replace("\r", "\\r")
    )
    return f'"{escaped}"'


# ---------- validated-set hashing ----------


def _validated_set_payload(vs: ValidatedTripleSet) -> dict[str, Any]:
    """Order-independent payload of ``vs`` for SHA derivation."""
    return {
        "subject_id": vs.subject_id,
        "schema_version": vs.schema_version,
        "concepts": sorted(
            (
                {
                    "canonical_term": c.canonical_term,
                    "definition": c.definition,
                    "sources": sorted(
                        (
                            {
                                "source_sha": s.source_sha,
                                "span": s.span,
                                "adapter": s.adapter,
                                "retrieved_at": s.retrieved_at,
                            }
                            for s in c.sources
                        ),
                        key=lambda d: (d["source_sha"], d["adapter"]),
                    ),
                }
                for c in vs.concepts
            ),
            key=lambda d: (d["canonical_term"],),
        ),
        "relations": sorted(
            (
                {
                    "head": r.head,
                    "relation": r.relation,
                    "tail": r.tail,
                    "sources": sorted(
                        (
                            {
                                "source_sha": s.source_sha,
                                "span": s.span,
                                "adapter": s.adapter,
                                "retrieved_at": s.retrieved_at,
                            }
                            for s in r.sources
                        ),
                        key=lambda d: (d["source_sha"], d["adapter"]),
                    ),
                }
                for r in vs.relations
            ),
            key=lambda d: (d["head"], d["relation"], d["tail"]),
        ),
        "counters": sorted(
            (
                {
                    "head": c.head,
                    "relation": c.relation,
                    "tail": c.tail,
                }
                for c in vs.counters
            ),
            key=lambda d: (d["head"], d["relation"], d["tail"]),
        ),
        "ordering_hints": sorted(
            (
                {"before": o.before, "after": o.after}
                for o in vs.ordering_hints
            ),
            key=lambda d: (d["before"], d["after"]),
        ),
    }


__all__ = ["compose"]