rfc-app/backend/app/metadata_schema.py

"""§22.4a configurable collection metadata — field schema + central validation.

SLICE-2 of docs/design/2026-06-06-configurable-collection-metadata.md.

A collection declares a small **field schema** in its `.collection.yaml`
(`fields:` block) so its entries can carry structured metadata — priority, tags,
and any custom fields the deployment defines. This module is the **one place**
that knows a collection's field shapes (modeled on `registry.py`):

  - `parse_fields` — normalize + validate the declared schema, leniently: a bad
    block or a bad field def is skipped with a warning, never raised, so a typo
    in one field can't nuke the collection mirror (INV-3 spirit). The normalized
    schema is a plain, JSON-serializable mapping that rides in
    `collections.config_json` (no DB migration) and is served verbatim by the
    collection API.
  - `validate` — check an entry's stored values against the schema, returning a
    list of advisory `Problem`s. Empty list = clean. Used **advisory at read**
    (the corpus mirror flags a non-empty result as `metadata_malformed`, INV-3)
    and is the enforcement point at the **write** boundary (the metadata-edit
    endpoints land in SLICE-4/5).

Field types (v1): `enum` (single scalar, controlled by a required `values:`
list), `tags` (a list; free-form unless `values:` given), `text` (a free
string). `ref` / `multi-enum` are future (design §2, Q2). Unknown types are
ignored with a warning. Keys an entry carries that the schema does **not**
declare ride along untouched and are never flagged (INV-7).
"""
from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import Any

log = logging.getLogger(__name__)

# §6.3 v1 field types. `ref` (typed cross-entry link) and `multi-enum` are
# deferred (design §2, Q2) — declared with an unknown type they're skipped.
VALID_FIELD_TYPES = {"enum", "tags", "text"}


@dataclass
class Problem:
    """One advisory schema-validation problem against a declared field.

    `code` is a stable machine token (`not-in-values`, `wrong-type`); `message`
    is human-facing (surfaced at the write boundary in SLICE-4/5)."""
    field: str
    code: str
    message: str

    def as_dict(self) -> dict[str, str]:
        return {"field": self.field, "code": self.code, "message": self.message}


# ----- schema parsing (lenient) -----

def parse_fields(raw: Any) -> dict[str, dict]:
    """Normalize a `.collection.yaml` `fields:` block → `{name: {type, ...}}`.

    Pure (no I/O). Lenient (INV-3): a non-mapping block yields `{}`; an
    individual field def that is not a mapping, has an unknown/missing `type`, or
    is an `enum` without a non-empty `values:` list is **skipped with a warning**
    — never raised. Order is preserved (facet display order, SLICE-3). The
    result is plain dicts so it serializes straight into `config_json` and the
    collection API.
    """
    if not isinstance(raw, dict):
        if raw is not None:
            log.warning("metadata_schema: fields block is not a mapping (%s); ignoring",
                        type(raw).__name__)
        return {}
    out: dict[str, dict] = {}
    for name, spec in raw.items():
        if not isinstance(spec, dict):
            log.warning("metadata_schema: field %r def is not a mapping; skipping", name)
            continue
        ftype = str(spec.get("type") or "").strip()
        if ftype not in VALID_FIELD_TYPES:
            log.warning("metadata_schema: field %r has unknown type %r; skipping",
                        name, ftype)
            continue
        values = spec.get("values")
        norm_values: list[str] | None = None
        if values is not None:
            if not isinstance(values, list):
                log.warning("metadata_schema: field %r values is not a list; ignoring",
                            name)
            else:
                norm_values = [str(v) for v in values]
        if ftype == "enum" and not norm_values:
            log.warning("metadata_schema: enum field %r needs a non-empty values "
                        "list; skipping", name)
            continue
        field_def: dict[str, Any] = {"type": ftype}
        if norm_values is not None:
            field_def["values"] = norm_values
        label = spec.get("label")
        if label:
            field_def["label"] = str(label)
        out[str(name)] = field_def
    return out


# ----- value validation (advisory) -----

def _is_scalar(v: Any) -> bool:
    return isinstance(v, (str, int, float, bool))


def validate(values: dict[str, Any], fields: dict[str, dict]) -> list[Problem]:
    """Check an entry's metadata `values` against a collection's field schema.

    Returns advisory `Problem`s (empty = clean). Only **declared** fields are
    checked; a field the entry omits is fine (no required fields in v1), and a
    key the schema doesn't declare rides along untouched (INV-7). With an empty
    schema, everything is clean (INV-5). Never raises (INV-3).
    """
    problems: list[Problem] = []
    for name, spec in fields.items():
        if name not in values:
            continue
        value = values[name]
        if value is None:
            continue
        ftype = spec.get("type")
        allowed = spec.get("values")
        if ftype == "enum":
            if not _is_scalar(value):
                problems.append(Problem(name, "wrong-type",
                    f"{name!r} must be a single value, got {type(value).__name__}"))
            elif allowed is not None and str(value) not in allowed:
                problems.append(Problem(name, "not-in-values",
                    f"{name!r} value {value!r} is not one of {allowed}"))
        elif ftype == "tags":
            if not isinstance(value, list):
                problems.append(Problem(name, "wrong-type",
                    f"{name!r} must be a list, got {type(value).__name__}"))
            elif allowed is not None:
                for member in value:
                    if str(member) not in allowed:
                        problems.append(Problem(name, "not-in-values",
                            f"{name!r} value {member!r} is not one of {allowed}"))
        elif ftype == "text":
            if not _is_scalar(value):
                problems.append(Problem(name, "wrong-type",
                    f"{name!r} must be a string, got {type(value).__name__}"))
    return problems