Parser

The parser turns OBML YAML into a semantic model, keeping source spans so errors point back at the line that caused them.

YAML Parser

orionbelt.parser.loader.TrackedLoader

YAML loader that tracks source positions for error reporting.

Uses ruamel.yaml which preserves line/column info on every parsed node.

Source code in src/orionbelt/parser/loader.py

class TrackedLoader:
    """YAML loader that tracks source positions for error reporting.

    Uses ruamel.yaml which preserves line/column info on every parsed node.
    """

    def __init__(self) -> None:
        self._yaml = YAML()
        self._yaml.preserve_quotes = True
        # Reject duplicate YAML keys (e.g. two columns with the same name).
        # Without this, ruamel.yaml silently keeps only the last value.
        self._yaml.allow_duplicate_keys = False
        # Reject deeply nested structures (mitigates stack-based DoS).
        # ruamel.yaml raises an error when nesting exceeds this limit.
        self._yaml.max_depth = _MAX_DEPTH

    # -- safety checks -------------------------------------------------------

    @staticmethod
    def _check_yaml_safety(content: str) -> None:
        """Pre-parse safety checks on raw YAML text.

        Raises ``YAMLSafetyError`` if the content contains anchors/aliases
        (not used in OBML) or exceeds the maximum document size.
        """
        if len(content) > _MAX_DOCUMENT_SIZE:
            raise YAMLSafetyError(
                f"YAML document exceeds maximum size "
                f"({len(content):,} chars > {_MAX_DOCUMENT_SIZE:,} limit)"
            )
        # Strip full-line comments before scanning so that &name inside
        # comments (e.g. "# see R&D notes") does not cause a false positive.
        stripped = _COMMENT_LINE_RE.sub("", content)
        if _ANCHOR_RE.search(stripped):
            raise YAMLSafetyError("YAML anchors/aliases are not supported in OBML")

    @staticmethod
    def _check_node_count(data: Any, limit: int = _MAX_NODE_COUNT) -> None:
        """Post-parse defense-in-depth: reject documents with too many nodes."""
        count = 0
        stack: list[Any] = [data]
        while stack:
            node = stack.pop()
            count += 1
            if count > limit:
                raise YAMLSafetyError(f"YAML document exceeds maximum node count ({limit:,})")
            if isinstance(node, dict):
                stack.extend(node.values())
            elif isinstance(node, list):
                stack.extend(node)

    # -- public loading API --------------------------------------------------

    def load(self, path: Path) -> tuple[dict[str, Any], SourceMap]:
        """Load a YAML file and return parsed dict + source position map."""
        with path.open("r", encoding="utf-8") as handle:
            content = handle.read()
        self._check_yaml_safety(content)
        data = self._yaml.load(content)
        if data is None:
            return {}, SourceMap()
        self._check_node_count(data)
        source_map = SourceMap()
        self._extract_positions(data, str(path), "", source_map)
        return self._to_plain_dict(data), source_map

    def load_string(
        self, content: str, filename: str = "<string>"
    ) -> tuple[dict[str, Any], SourceMap]:
        """Load YAML from a string."""
        self._check_yaml_safety(content)
        data = self._yaml.load(content)
        if data is None:
            return {}, SourceMap()
        self._check_node_count(data)
        source_map = SourceMap()
        self._extract_positions(data, filename, "", source_map)
        return self._to_plain_dict(data), source_map

    def _extract_positions(
        self,
        data: Any,
        filename: str,
        prefix: str,
        source_map: SourceMap,
    ) -> None:
        """Recursively extract source positions from ruamel.yaml nodes."""
        if isinstance(data, CommentedMap):
            for key in data:
                key_path = f"{prefix}.{key}" if prefix else str(key)
                # Try to get position for this key from ruamel.yaml's lc object
                try:
                    lc = data.lc
                    # lc.key() returns a callable in newer ruamel.yaml
                    key_positions = lc.key(key)
                    if key_positions:
                        line, col = key_positions
                        source_map.add(
                            key_path,
                            SourceSpan(file=filename, line=line + 1, column=col + 1),
                        )
                except (AttributeError, KeyError, TypeError):
                    # Fallback: use the map's own position
                    try:
                        lc = data.lc
                        source_map.add(
                            key_path,
                            SourceSpan(file=filename, line=lc.line + 1, column=lc.col + 1),
                        )
                    except (AttributeError, TypeError):
                        pass
                self._extract_positions(data[key], filename, key_path, source_map)
        elif isinstance(data, CommentedSeq):
            for i, item in enumerate(data):
                item_path = f"{prefix}[{i}]"
                try:
                    lc = data.lc
                    item_pos = lc.item(i)
                    if item_pos:
                        line, col = item_pos
                        source_map.add(
                            item_path,
                            SourceSpan(file=filename, line=line + 1, column=col + 1),
                        )
                except (AttributeError, KeyError, TypeError):
                    pass
                self._extract_positions(item, filename, item_path, source_map)

    def _to_plain_dict(self, data: Any) -> dict[str, Any]:
        """Convert ruamel.yaml CommentedMap/Seq to plain Python dict/list."""
        if isinstance(data, CommentedMap):
            return {str(k): self._to_plain_value(v) for k, v in data.items()}
        if isinstance(data, dict):
            return {str(k): self._to_plain_value(v) for k, v in data.items()}
        return {}

    def _to_plain_value(self, data: Any) -> Any:
        if isinstance(data, CommentedMap):
            return {str(k): self._to_plain_value(v) for k, v in data.items()}
        if isinstance(data, CommentedSeq):
            return [self._to_plain_value(item) for item in data]
        if isinstance(data, dict):
            return {str(k): self._to_plain_value(v) for k, v in data.items()}
        if isinstance(data, list):
            return [self._to_plain_value(item) for item in data]
        return data

load(path)

Load a YAML file and return parsed dict + source position map.

Source code in src/orionbelt/parser/loader.py

def load(self, path: Path) -> tuple[dict[str, Any], SourceMap]:
    """Load a YAML file and return parsed dict + source position map."""
    with path.open("r", encoding="utf-8") as handle:
        content = handle.read()
    self._check_yaml_safety(content)
    data = self._yaml.load(content)
    if data is None:
        return {}, SourceMap()
    self._check_node_count(data)
    source_map = SourceMap()
    self._extract_positions(data, str(path), "", source_map)
    return self._to_plain_dict(data), source_map

load_string(content, filename='<string>')

Load YAML from a string.

Source code in src/orionbelt/parser/loader.py

def load_string(
    self, content: str, filename: str = "<string>"
) -> tuple[dict[str, Any], SourceMap]:
    """Load YAML from a string."""
    self._check_yaml_safety(content)
    data = self._yaml.load(content)
    if data is None:
        return {}, SourceMap()
    self._check_node_count(data)
    source_map = SourceMap()
    self._extract_positions(data, filename, "", source_map)
    return self._to_plain_dict(data), source_map

Reference Resolver

orionbelt.parser.resolver.ReferenceResolver

Resolves all references in a raw YAML model to a fully-typed SemanticModel.

Source code in src/orionbelt/parser/resolver.py

class ReferenceResolver:
    """Resolves all references in a raw YAML model to a fully-typed SemanticModel."""

    def resolve(
        self,
        raw: dict[str, Any],
        source_map: SourceMap | None = None,
    ) -> tuple[SemanticModel, ValidationResult]:
        """Resolve raw YAML dict into a validated SemanticModel.

        Returns (model, validation_result). If there are errors,
        the model may be partially populated.
        """
        errors: list[SemanticError] = []
        warnings: list[SemanticError] = []

        # Strict OBML: reject unknown top-level keys (catches typos like
        # ``dataObjekt:`` that would silently be dropped by ``raw.get(...)``).
        _check_unknown_keys(raw, _TOP_LEVEL_KEYS, "", errors, source_map)

        # Ontology prefixes come first: every object's externalConceptMappings
        # expand against them.
        ontology = _parse_ontology(raw.get("ontology"), errors, source_map)
        prefixes = effective_prefixes(ontology.prefixes if ontology else None)

        # Parse data objects
        data_objects: dict[str, DataObject] = {}
        raw_objects = raw.get("dataObjects", {})
        if not isinstance(raw_objects, dict):
            errors.append(
                SemanticError(
                    code="DATA_OBJECT_PARSE_ERROR",
                    message="'dataObjects' must be a YAML mapping, not a list or scalar",
                    path="dataObjects",
                )
            )
            raw_objects = {}
        for name, raw_obj in raw_objects.items():
            try:
                _check_unknown_keys(
                    raw_obj, _DATA_OBJECT_KEYS, f"dataObjects.{name}", errors, source_map
                )
                obj_columns: dict[str, DataObjectColumn] = {}
                for fname, fdata in raw_obj.get("columns", {}).items():
                    _check_unknown_keys(
                        fdata,
                        _DATA_OBJECT_COLUMN_KEYS,
                        f"dataObjects.{name}.columns.{fname}",
                        errors,
                        source_map,
                    )
                    obj_columns[fname] = DataObjectColumn(
                        name=fname,
                        code=fdata.get("code", fname if not fdata.get("expression") else ""),
                        abstract_type=fdata.get("abstractType", "string"),
                        sql_type=fdata.get("sqlType"),
                        sql_precision=fdata.get("sqlPrecision"),
                        sql_scale=fdata.get("sqlScale"),
                        num_class=fdata.get("numClass"),
                        primary_key=bool(fdata.get("primaryKey", False)),
                        description=fdata.get("description"),
                        comment=fdata.get("comment"),
                        owner=fdata.get("owner"),
                        expression=fdata.get("expression"),
                        synonyms=fdata.get("synonyms", []),
                        custom_extensions=_parse_extensions(fdata),
                    )

                obj_joins: list[DataObjectJoin] = []
                for ji, jdata in enumerate(raw_obj.get("joins", [])):
                    _check_unknown_keys(
                        jdata,
                        _DATA_OBJECT_JOIN_KEYS,
                        f"dataObjects.{name}.joins[{ji}]",
                        errors,
                        source_map,
                    )
                    obj_joins.append(
                        DataObjectJoin(
                            join_type=jdata["joinType"],
                            join_to=jdata["joinTo"],
                            columns_from=jdata["columnsFrom"],
                            columns_to=jdata["columnsTo"],
                            secondary=jdata.get("secondary", False),
                            path_name=jdata.get("pathName"),
                            required=jdata.get("required", False),
                        )
                    )

                data_objects[name] = DataObject(
                    name=name,
                    code=raw_obj.get("code", ""),
                    database=raw_obj.get("database", ""),
                    schema_name=raw_obj.get("schema", ""),
                    columns=obj_columns,
                    joins=obj_joins,
                    description=raw_obj.get("description"),
                    comment=raw_obj.get("comment"),
                    owner=raw_obj.get("owner"),
                    countable=raw_obj.get("countable", True),
                    count_label=raw_obj.get("countLabel"),
                    synonyms=raw_obj.get("synonyms", []),
                    custom_extensions=_parse_extensions(raw_obj),
                    external_concept_mappings=_parse_concept_mappings(
                        raw_obj, f"dataObjects.{name}", prefixes, errors, source_map
                    ),
                    refresh=_parse_refresh(raw_obj.get("refresh"), name, errors),
                    nested_in=_parse_nested_in(raw_obj.get("nestedIn")),
                )
            except Exception as e:
                span = source_map.get(f"dataObjects.{name}") if source_map else None
                errors.append(
                    SemanticError(
                        code="DATA_OBJECT_PARSE_ERROR",
                        message=f"Failed to parse data object '{name}': {e}",
                        path=f"dataObjects.{name}",
                        span=span,
                    )
                )

        # Parse dimensions
        dimensions: dict[str, Dimension] = {}
        raw_dims = raw.get("dimensions", {})
        if not isinstance(raw_dims, dict):
            errors.append(
                SemanticError(
                    code="DIMENSION_PARSE_ERROR",
                    message="'dimensions' must be a YAML mapping, not a list or scalar",
                    path="dimensions",
                )
            )
            raw_dims = {}
        for name, raw_dim in raw_dims.items():
            try:
                _check_unknown_keys(
                    raw_dim, _DIMENSION_KEYS, f"dimensions.{name}", errors, source_map
                )
                data_object = raw_dim.get("dataObject")
                column = raw_dim.get("column")

                # Validate the data object exists
                if data_object and data_object not in data_objects:
                    span = source_map.get(f"dimensions.{name}") if source_map else None
                    errors.append(
                        SemanticError(
                            code="UNKNOWN_DATA_OBJECT",
                            message=(
                                f"Dimension '{name}' references unknown data object '{data_object}'"
                            ),
                            path=f"dimensions.{name}",
                            span=span,
                            suggestions=_suggest_similar(data_object, list(data_objects.keys())),
                        )
                    )

                # Validate the column exists in the data object
                if (
                    data_object
                    and column
                    and data_object in data_objects
                    and column not in data_objects[data_object].columns
                ):
                    span = source_map.get(f"dimensions.{name}") if source_map else None
                    errors.append(
                        SemanticError(
                            code="UNKNOWN_COLUMN",
                            message=(
                                f"Dimension '{name}' references unknown column "
                                f"'{column}' in data object '{data_object}'"
                            ),
                            path=f"dimensions.{name}",
                            span=span,
                            suggestions=_suggest_similar(
                                column, list(data_objects[data_object].columns.keys())
                            ),
                        )
                    )

                via = raw_dim.get("via")
                if via and via not in data_objects:
                    span = source_map.get(f"dimensions.{name}") if source_map else None
                    errors.append(
                        SemanticError(
                            code="UNKNOWN_DATA_OBJECT",
                            message=(
                                f"Dimension '{name}' via references unknown data object '{via}'"
                            ),
                            path=f"dimensions.{name}",
                            span=span,
                            suggestions=_suggest_similar(via, list(data_objects.keys())),
                        )
                    )

                dimensions[name] = Dimension(
                    name=name,
                    view=data_object or "",
                    column=column or "",
                    result_type=raw_dim.get("resultType", "string"),
                    time_grain=raw_dim.get("timeGrain"),
                    via=via,
                    path_name=raw_dim.get("pathName"),
                    description=raw_dim.get("description"),
                    format=raw_dim.get("format"),
                    owner=raw_dim.get("owner"),
                    synonyms=raw_dim.get("synonyms", []),
                    custom_extensions=_parse_extensions(raw_dim),
                    external_concept_mappings=_parse_concept_mappings(
                        raw_dim, f"dimensions.{name}", prefixes, errors, source_map
                    ),
                )
            except Exception as e:
                span = source_map.get(f"dimensions.{name}") if source_map else None
                errors.append(
                    SemanticError(
                        code="DIMENSION_PARSE_ERROR",
                        message=f"Failed to parse dimension '{name}': {e}",
                        path=f"dimensions.{name}",
                        span=span,
                    )
                )

        # Parse measures
        measures: dict[str, Measure] = {}
        raw_measures = raw.get("measures", {})
        if not isinstance(raw_measures, dict):
            errors.append(
                SemanticError(
                    code="MEASURE_PARSE_ERROR",
                    message="'measures' must be a YAML mapping, not a list or scalar",
                    path="measures",
                )
            )
            raw_measures = {}
        for name, raw_meas in raw_measures.items():
            try:
                _check_unknown_keys(raw_meas, _MEASURE_KEYS, f"measures.{name}", errors, source_map)
                measure_columns: list[DataColumnRef] = []
                for ci, fdata in enumerate(raw_meas.get("columns", [])):
                    _check_unknown_keys(
                        fdata,
                        _DATA_COLUMN_REF_KEYS,
                        f"measures.{name}.columns[{ci}]",
                        errors,
                        source_map,
                    )
                    measure_columns.append(
                        DataColumnRef(
                            view=fdata.get("dataObject"),
                            column=fdata.get("column"),
                        )
                    )

                # Resolve expression field references
                expression = raw_meas.get("expression")
                if expression:
                    self._validate_expression_refs(
                        name, expression, data_objects, errors, source_map
                    )

                # Parse measure filters (new `filters:` list or legacy `filter:` single)
                measure_filters: list[MeasureFilterItem] = []
                raw_filters = raw_meas.get("filters")
                if raw_filters and isinstance(raw_filters, list):
                    for fi, rf in enumerate(raw_filters):
                        measure_filters.append(
                            _parse_measure_filter_item(
                                rf,
                                f"measures.{name}.filters[{fi}]",
                                errors,
                                source_map,
                            )
                        )
                else:
                    # Backward compat: single `filter:` key → [filter]
                    raw_filter = raw_meas.get("filter")
                    if raw_filter:
                        measure_filters.append(
                            _parse_measure_filter_item(
                                raw_filter, f"measures.{name}.filter", errors, source_map
                            )
                        )

                # Parse grain override
                grain_override: GrainOverride | None = None
                raw_grain = raw_meas.get("grain")
                if raw_grain and isinstance(raw_grain, dict):
                    _check_unknown_keys(
                        raw_grain,
                        _GRAIN_OVERRIDE_KEYS,
                        f"measures.{name}.grain",
                        errors,
                        source_map,
                    )
                    grain_override = GrainOverride(
                        mode=raw_grain.get("mode", "RELATIVE"),
                        exclude=raw_grain.get("exclude", []),
                        include=raw_grain.get("include", []),
                        keep_only=raw_grain.get("keepOnly", []),
                    )
                    # Validate dimension references in grain
                    for dim_name in (
                        grain_override.include + grain_override.exclude + grain_override.keep_only
                    ):
                        if dim_name not in dimensions:
                            span = source_map.get(f"measures.{name}.grain") if source_map else None
                            errors.append(
                                SemanticError(
                                    code="UNKNOWN_GRAIN_DIMENSION",
                                    message=(
                                        f"Measure '{name}' grain references "
                                        f"unknown dimension '{dim_name}'"
                                    ),
                                    path=f"measures.{name}.grain",
                                    span=span,
                                    suggestions=_suggest_similar(dim_name, list(dimensions.keys())),
                                )
                            )

                # Parse filter context
                filter_ctx: FilterContext | None = None
                raw_fc = raw_meas.get("filterContext")
                if raw_fc and isinstance(raw_fc, dict):
                    _check_unknown_keys(
                        raw_fc,
                        _FILTER_CONTEXT_KEYS,
                        f"measures.{name}.filterContext",
                        errors,
                        source_map,
                    )
                    include_filters: list[FilterContextFilter] = []
                    for inc_i, raw_incl in enumerate(raw_fc.get("include", [])):
                        if isinstance(raw_incl, dict):
                            _check_unknown_keys(
                                raw_incl,
                                _FILTER_CONTEXT_FILTER_KEYS,
                                f"measures.{name}.filterContext.include[{inc_i}]",
                                errors,
                                source_map,
                            )
                            include_filters.append(
                                FilterContextFilter(
                                    field=raw_incl.get("field", ""),
                                    op=raw_incl.get("op", "equals"),
                                    value=raw_incl.get("value"),
                                )
                            )
                    filter_ctx = FilterContext(
                        mode=raw_fc.get("mode", "RELATIVE"),
                        exclude=raw_fc.get("exclude", []),
                        include=include_filters,
                        keep_only=raw_fc.get("keepOnly", []),
                    )
                    # Validate field references in exclude/keepOnly
                    all_dim_names = set(dimensions.keys())
                    all_col_refs: set[str] = set()
                    for obj_name, obj_def in data_objects.items():
                        for col_name in obj_def.columns:
                            all_col_refs.add(f"{obj_name}.{col_name}")
                    for field_name in filter_ctx.exclude + filter_ctx.keep_only:
                        if field_name not in all_dim_names and field_name not in all_col_refs:
                            span = (
                                source_map.get(f"measures.{name}.filterContext")
                                if source_map
                                else None
                            )
                            errors.append(
                                SemanticError(
                                    code="UNKNOWN_FILTER_CONTEXT_FIELD",
                                    message=(
                                        f"Measure '{name}' filterContext references "
                                        f"unknown field '{field_name}'"
                                    ),
                                    path=f"measures.{name}.filterContext",
                                    span=span,
                                    suggestions=_suggest_similar(field_name, list(all_dim_names)),
                                )
                            )
                    for incl in filter_ctx.include:
                        if incl.field not in all_dim_names and incl.field not in all_col_refs:
                            span = (
                                source_map.get(f"measures.{name}.filterContext")
                                if source_map
                                else None
                            )
                            errors.append(
                                SemanticError(
                                    code="UNKNOWN_FILTER_CONTEXT_FIELD",
                                    message=(
                                        f"Measure '{name}' filterContext.include "
                                        f"references unknown field '{incl.field}'"
                                    ),
                                    path=f"measures.{name}.filterContext.include",
                                    span=span,
                                    suggestions=_suggest_similar(incl.field, list(all_dim_names)),
                                )
                            )

                measures[name] = Measure(
                    name=name,
                    columns=measure_columns,
                    result_type=raw_meas.get("resultType", "float"),
                    aggregation=raw_meas.get("aggregation", "sum"),
                    expression=expression,
                    distinct=raw_meas.get("distinct", False),
                    total=raw_meas.get("total", False),
                    default_value=raw_meas.get("defaultValue"),
                    anchor=raw_meas.get("anchor"),
                    grain=grain_override,
                    filter_context=filter_ctx,
                    filters=measure_filters,
                    data_type=raw_meas.get("dataType"),
                    description=raw_meas.get("description"),
                    format=raw_meas.get("format"),
                    allow_fan_out=raw_meas.get("allowFanOut", False),
                    delimiter=raw_meas.get("delimiter"),
                    percentile=raw_meas.get("percentile"),
                    within_group=raw_meas.get("withinGroup"),
                    owner=raw_meas.get("owner"),
                    synonyms=raw_meas.get("synonyms", []),
                    custom_extensions=_parse_extensions(raw_meas),
                    external_concept_mappings=_parse_concept_mappings(
                        raw_meas, f"measures.{name}", prefixes, errors, source_map
                    ),
                )
            except Exception as e:
                span = source_map.get(f"measures.{name}") if source_map else None
                errors.append(
                    SemanticError(
                        code="MEASURE_PARSE_ERROR",
                        message=f"Failed to parse measure '{name}': {e}",
                        path=f"measures.{name}",
                        span=span,
                    )
                )

        # Validate the count-synthesis knobs here so a bad value becomes a
        # structured SemanticError rather than a raw AttributeError (list
        # pattern) or an uncaught Pydantic ValidationError (invalid token) at
        # model construction below. Fall back to safe values so resolution can
        # continue collecting errors.
        _count_pattern = raw.get("countLabelPattern", DEFAULT_COUNT_PATTERN)
        _pattern_err = count_pattern_error(_count_pattern)
        if _pattern_err is not None:
            span = source_map.get("countLabelPattern") if source_map else None
            errors.append(
                SemanticError(
                    code="INVALID_COUNT_LABEL_PATTERN",
                    message=_pattern_err,
                    path="countLabelPattern",
                    span=span,
                )
            )
            _count_pattern = DEFAULT_COUNT_PATTERN
        _expose_counts = raw.get("exposeCounts", True)
        if not isinstance(_expose_counts, bool):
            span = source_map.get("exposeCounts") if source_map else None
            errors.append(
                SemanticError(
                    code="INVALID_EXPOSE_COUNTS",
                    message="exposeCounts must be a boolean (true/false)",
                    path="exposeCounts",
                    span=span,
                )
            )
            _expose_counts = True

        # Names of synthesized count measures (name == resolved count label,
        # e.g. "Sales Count"). These are valid measure references (metrics may
        # target them) even though they are not declared — they are materialized
        # on read via ``effective_measures`` (see models/synthesis.py). Declared
        # measures already sit in ``measures``; a declared count of the same
        # name overrides synthesis, so unioning is safe either way.
        synthesized_measure_names: set[str] = (
            {
                count_label(key, obj, _count_pattern)
                for key, obj in data_objects.items()
                if obj.countable
            }
            if _expose_counts
            else set()
        )

        # Parse metrics
        metrics: dict[str, Metric] = {}
        raw_metrics = raw.get("metrics", {})
        if not isinstance(raw_metrics, dict):
            errors.append(
                SemanticError(
                    code="METRIC_PARSE_ERROR",
                    message="'metrics' must be a YAML mapping, not a list or scalar",
                    path="metrics",
                )
            )
            raw_metrics = {}
        for name, raw_metric in raw_metrics.items():
            try:
                _check_unknown_keys(raw_metric, _METRIC_KEYS, f"metrics.{name}", errors, source_map)
                raw_pop_block = raw_metric.get("periodOverPeriod")
                if isinstance(raw_pop_block, dict):
                    _check_unknown_keys(
                        raw_pop_block,
                        _PERIOD_OVER_PERIOD_KEYS,
                        f"metrics.{name}.periodOverPeriod",
                        errors,
                        source_map,
                    )
                metric_type = raw_metric.get("type", "derived")
                # The other branches pass only their own fields to the model, so
                # these would be dropped silently rather than refused there.
                if metric_type != MetricType.REAGGREGATE and any(
                    key in raw_metric
                    for key in (
                        "per",
                        "aggregation",
                        "having",
                        "percentile",
                        "population",
                        "carryForward",
                    )
                ):
                    raise ValueError(
                        "'per', 'aggregation', 'having', 'percentile', 'population' and "
                        "'carryForward' are only valid on reaggregate metrics"
                    )
                concept_links = _parse_concept_mappings(
                    raw_metric, f"metrics.{name}", prefixes, errors, source_map
                )

                if metric_type == MetricType.CUMULATIVE:
                    # Cumulative metric: validate measure reference exists
                    ref_measure = raw_metric.get("measure", "")
                    if (
                        ref_measure
                        and ref_measure not in measures
                        and ref_measure not in synthesized_measure_names
                        and not _is_reaggregate(raw_metrics.get(ref_measure))
                    ):
                        span = source_map.get(f"metrics.{name}.measure") if source_map else None
                        errors.append(
                            SemanticError(
                                code="UNKNOWN_MEASURE",
                                message=(
                                    f"Cumulative metric '{name}' references "
                                    f"unknown measure '{ref_measure}'"
                                ),
                                path=f"metrics.{name}.measure",
                                span=span,
                            )
                        )

                    # Validate timeDimension references a known dimension
                    cum_time_dim = raw_metric.get("timeDimension", "")
                    if cum_time_dim and cum_time_dim not in dimensions:
                        span = (
                            source_map.get(f"metrics.{name}.timeDimension") if source_map else None
                        )
                        errors.append(
                            SemanticError(
                                code="CUMULATIVE_UNKNOWN_TIME_DIMENSION",
                                message=(
                                    f"Cumulative metric '{name}' references "
                                    f"unknown time dimension '{cum_time_dim}'"
                                ),
                                path=f"metrics.{name}.timeDimension",
                                span=span,
                                suggestions=_suggest_similar(cum_time_dim, list(dimensions.keys())),
                            )
                        )

                    metrics[name] = Metric(
                        name=name,
                        type=MetricType.CUMULATIVE,
                        measure=raw_metric.get("measure"),
                        time_dimension=raw_metric.get("timeDimension"),
                        cumulative_type=raw_metric.get("cumulativeType", "sum"),
                        window=raw_metric.get("window"),
                        grain_to_date=raw_metric.get("grainToDate"),
                        partition_by=list(raw_metric.get("partitionBy", []) or []),
                        data_type=raw_metric.get("dataType"),
                        description=raw_metric.get("description"),
                        format=raw_metric.get("format"),
                        owner=raw_metric.get("owner"),
                        synonyms=raw_metric.get("synonyms", []),
                        custom_extensions=_parse_extensions(raw_metric),
                        external_concept_mappings=concept_links,
                    )
                elif metric_type == MetricType.PERIOD_OVER_PERIOD:
                    # Period-over-period metric: validate expression + PoP config.
                    # Its base has to be re-aggregated per period from the fact
                    # tables, so the reference must be a measure.
                    expression = raw_metric.get("expression", "")
                    self._validate_metric_expression_refs(
                        name,
                        expression,
                        measures,
                        errors,
                        source_map,
                        metrics,
                        synthesized_measure_names,
                    )

                    raw_pop = raw_metric.get("periodOverPeriod")
                    if not raw_pop:
                        span = source_map.get(f"metrics.{name}") if source_map else None
                        errors.append(
                            SemanticError(
                                code="METRIC_PARSE_ERROR",
                                message=(
                                    f"Period-over-period metric '{name}' "
                                    f"requires 'periodOverPeriod' configuration"
                                ),
                                path=f"metrics.{name}",
                                span=span,
                            )
                        )
                        raw_pop = {}

                    # Validate time dimension reference
                    pop_time_dim = raw_pop.get("timeDimension", "")
                    if pop_time_dim and pop_time_dim not in dimensions:
                        span = (
                            source_map.get(f"metrics.{name}.periodOverPeriod")
                            if source_map
                            else None
                        )
                        errors.append(
                            SemanticError(
                                code="POP_UNKNOWN_TIME_DIMENSION",
                                message=(
                                    f"Period-over-period metric '{name}' references "
                                    f"unknown time dimension '{pop_time_dim}'"
                                ),
                                path=f"metrics.{name}.periodOverPeriod.timeDimension",
                                span=span,
                                suggestions=_suggest_similar(pop_time_dim, list(dimensions.keys())),
                            )
                        )

                    pop_config = PeriodOverPeriod(
                        time_dimension=raw_pop.get("timeDimension", ""),
                        grain=raw_pop.get("grain", "month"),
                        offset=raw_pop.get("offset", -1),
                        offset_grain=raw_pop.get("offsetGrain", "year"),
                        comparison=raw_pop.get("comparison", "percentChange"),
                    )

                    metrics[name] = Metric(
                        name=name,
                        type=MetricType.PERIOD_OVER_PERIOD,
                        expression=expression,
                        period_over_period=pop_config,
                        data_type=raw_metric.get("dataType"),
                        description=raw_metric.get("description"),
                        format=raw_metric.get("format"),
                        owner=raw_metric.get("owner"),
                        synonyms=raw_metric.get("synonyms", []),
                        custom_extensions=_parse_extensions(raw_metric),
                        external_concept_mappings=concept_links,
                    )
                elif metric_type == MetricType.REAGGREGATE:
                    self._check_reaggregate_refs(
                        name,
                        raw_metric,
                        measures,
                        synthesized_measure_names,
                        raw_metrics,
                        dimensions,
                        errors,
                        source_map,
                    )
                    # Every recognised key is passed on, so a field of another
                    # metric type is refused by the model instead of dropped.
                    metrics[name] = Metric.model_validate(
                        {
                            **{
                                key: value
                                for key, value in raw_metric.items()
                                if key in _METRIC_KEYS
                                and key not in ("customExtensions", "externalConceptMappings")
                            },
                            "name": name,
                            "customExtensions": _parse_extensions(raw_metric),
                            "externalConceptMappings": concept_links,
                        }
                    )
                elif metric_type == MetricType.WINDOW:
                    # Window metric (rank/lag/lead/ntile/first_value/last_value)
                    ref_measure = raw_metric.get("measure")
                    # A reaggregate metric is a base too: the window then runs
                    # over its finished values (``compiler/late_window.py``).
                    if (
                        ref_measure
                        and ref_measure not in measures
                        and ref_measure not in synthesized_measure_names
                        and not _is_reaggregate(raw_metrics.get(ref_measure))
                    ):
                        span = source_map.get(f"metrics.{name}.measure") if source_map else None
                        errors.append(
                            SemanticError(
                                code="UNKNOWN_MEASURE",
                                message=(
                                    f"Window metric '{name}' references "
                                    f"unknown measure '{ref_measure}'"
                                ),
                                path=f"metrics.{name}.measure",
                                span=span,
                            )
                        )

                    win_time_dim = raw_metric.get("timeDimension", "")
                    if win_time_dim and win_time_dim not in dimensions:
                        span = (
                            source_map.get(f"metrics.{name}.timeDimension") if source_map else None
                        )
                        errors.append(
                            SemanticError(
                                code="WINDOW_UNKNOWN_TIME_DIMENSION",
                                message=(
                                    f"Window metric '{name}' references "
                                    f"unknown time dimension '{win_time_dim}'"
                                ),
                                path=f"metrics.{name}.timeDimension",
                                span=span,
                                suggestions=_suggest_similar(win_time_dim, list(dimensions.keys())),
                            )
                        )

                    metrics[name] = Metric(
                        name=name,
                        type=MetricType.WINDOW,
                        measure=ref_measure,
                        time_dimension=raw_metric.get("timeDimension"),
                        window_function=raw_metric.get("windowFunction"),
                        offset=raw_metric.get("offset"),
                        buckets=raw_metric.get("buckets"),
                        order_direction=raw_metric.get("orderDirection", "desc"),
                        default_value=raw_metric.get("defaultValue"),
                        partition_by=list(raw_metric.get("partitionBy", []) or []),
                        data_type=raw_metric.get("dataType"),
                        description=raw_metric.get("description"),
                        format=raw_metric.get("format"),
                        owner=raw_metric.get("owner"),
                        synonyms=raw_metric.get("synonyms", []),
                        custom_extensions=_parse_extensions(raw_metric),
                        external_concept_mappings=concept_links,
                    )
                else:
                    # Derived metric (default). It may reference another derived
                    # metric, expanded in place down to real aggregates, or a
                    # window metric, which the window wrapper projects as a
                    # column of its base CTE — that is what makes
                    # ``{[Revenue]} - {[Revenue Prior Month]}`` work, or a
                    # reaggregate metric, which reaggregate_wrap rebuilds the
                    # formula over.
                    expression = raw_metric.get("expression", "")
                    self._validate_metric_expression_refs(
                        name,
                        expression,
                        measures,
                        errors,
                        source_map,
                        metrics,
                        synthesized_measure_names,
                        composable_metric_types=(
                            MetricType.DERIVED,
                            MetricType.WINDOW,
                            MetricType.REAGGREGATE,
                        ),
                    )

                    metrics[name] = Metric(
                        name=name,
                        expression=expression,
                        data_type=raw_metric.get("dataType"),
                        description=raw_metric.get("description"),
                        format=raw_metric.get("format"),
                        owner=raw_metric.get("owner"),
                        synonyms=raw_metric.get("synonyms", []),
                        custom_extensions=_parse_extensions(raw_metric),
                        external_concept_mappings=concept_links,
                    )
            except Exception as e:
                span = source_map.get(f"metrics.{name}") if source_map else None
                errors.append(
                    SemanticError(
                        code="METRIC_PARSE_ERROR",
                        message=f"Failed to parse metric '{name}': {e}",
                        path=f"metrics.{name}",
                        span=span,
                    )
                )

        # Parse static model filters
        model_filters: list[ModelFilter] = []
        raw_filters = raw.get("filters", [])
        if not isinstance(raw_filters, list):
            errors.append(
                SemanticError(
                    code="FILTER_PARSE_ERROR",
                    message="'filters' must be a YAML list, not a mapping or scalar",
                    path="filters",
                )
            )
            raw_filters = []
        for i, rf in enumerate(raw_filters):
            try:
                _check_unknown_keys(rf, _MODEL_FILTER_KEYS, f"filters[{i}]", errors, source_map)
                obj_name = rf.get("dataObject", "")
                col_name = rf.get("column", "")
                if obj_name and obj_name not in data_objects:
                    span = source_map.get(f"filters[{i}]") if source_map else None
                    errors.append(
                        SemanticError(
                            code="UNKNOWN_FILTER_DATA_OBJECT",
                            message=(
                                f"Static filter[{i}] references unknown data object '{obj_name}'"
                            ),
                            path=f"filters[{i}]",
                            span=span,
                        )
                    )
                elif obj_name and col_name and col_name not in data_objects[obj_name].columns:
                    span = source_map.get(f"filters[{i}]") if source_map else None
                    errors.append(
                        SemanticError(
                            code="UNKNOWN_FILTER_COLUMN",
                            message=(
                                f"Static filter[{i}] references unknown column "
                                f"'{col_name}' in data object '{obj_name}'"
                            ),
                            path=f"filters[{i}]",
                            span=span,
                        )
                    )
                raw_val = rf.get("value")
                raw_vals = rf.get("values", [])
                model_filters.append(
                    ModelFilter(
                        data_object=obj_name,
                        column=col_name,
                        operator=rf.get("operator", "equals"),
                        value=_coerce_filter_value(raw_val),
                        values=[_coerce_filter_value(v) for v in raw_vals],
                    )
                )
            except Exception as e:
                span = source_map.get(f"filters[{i}]") if source_map else None
                errors.append(
                    SemanticError(
                        code="FILTER_PARSE_ERROR",
                        message=f"Failed to parse static filter[{i}]: {e}",
                        path=f"filters[{i}]",
                        span=span,
                    )
                )

        settings = _parse_settings(raw.get("settings"), errors, source_map)

        rules = self._parse_rules(
            raw.get("rules"),
            dimensions,
            set(measures) | synthesized_measure_names,
            set(metrics),
            prefixes,
            errors,
            source_map,
        )

        # Parse examples block (PLAN_agent_api_improvements §5)
        examples = self._parse_examples(raw.get("examples"), errors)

        model = SemanticModel(
            version=raw.get("version", 1.0),
            name=raw.get("name"),
            description=raw.get("description"),
            data_objects=data_objects,
            dimensions=dimensions,
            measures=measures,
            metrics=metrics,
            filters=model_filters,
            rules=rules,
            examples=examples,
            extends_sources=raw.get("_extends_sources", []),
            inherits_source=raw.get("_inherits_source"),
            owner=raw.get("owner"),
            # Sanitized above so an invalid value is a structured error, not a
            # ValidationError raised here.
            expose_counts=_expose_counts,
            count_label_pattern=_count_pattern,
            custom_extensions=_parse_extensions(raw, "", errors, source_map),
            ontology=ontology,
            external_concept_mappings=_parse_concept_mappings(
                raw, "", prefixes, errors, source_map
            ),
            settings=settings,
        )

        result = ValidationResult(
            valid=len(errors) == 0,
            errors=errors,
            warnings=warnings,
        )

        return model, result

    def _parse_rules(
        self,
        raw: object,
        dimensions: dict[str, Dimension],
        measure_names: set[str],
        metric_names: set[str],
        prefixes: dict[str, str],
        errors: list[SemanticError],
        source_map: SourceMap | None,
    ) -> dict[str, Rule]:
        """Parse and validate the ``rules:`` block.

        Two passes: build every rule first, then check the cross-rule facts
        (level, grain, references, cycles), since a rule's level depends on
        what the rules it references read. A rule with a bad condition is
        reported and dropped; the others still load.
        """
        if raw is None:
            return {}
        if not isinstance(raw, dict):
            errors.append(
                SemanticError(
                    code="RULE_PARSE_ERROR",
                    message="'rules' must be a YAML mapping keyed by rule name",
                    path="rules",
                    span=source_map.get("rules") if source_map else None,
                )
            )
            return {}
        aggregate_names = measure_names | metric_names
        known_fields = set(dimensions) | aggregate_names
        rules: dict[str, Rule] = {}
        for name, raw_rule in raw.items():
            path = f"rules.{name}"
            span = source_map.get(path) if source_map else None
            if not isinstance(raw_rule, dict):
                errors.append(
                    SemanticError(
                        code="RULE_PARSE_ERROR",
                        message=f"Rule '{name}' must be a mapping with a 'condition'",
                        path=path,
                        span=span,
                    )
                )
                continue
            _check_unknown_keys(raw_rule, _RULE_KEYS, path, errors, source_map)
            self._check_condition_keys(raw_rule.get("condition"), f"{path}.condition", errors)
            # The two list-valued blocks have their own parsers (unknown-key
            # checks, IRI expansion); everything else goes to the model as is.
            fields = {
                k: v
                for k, v in raw_rule.items()
                if k in _RULE_KEYS and k not in ("customExtensions", "externalConceptMappings")
            }
            try:
                rule = Rule(
                    name=name,
                    **fields,
                    custom_extensions=_parse_extensions(raw_rule),
                    external_concept_mappings=_parse_concept_mappings(
                        raw_rule, path, prefixes, errors, source_map
                    ),
                )
            except PydanticValidationError as exc:
                for detail in exc.errors():
                    loc = ".".join(str(part) for part in detail["loc"])
                    code = (
                        "INVALID_RULE_CONDITION"
                        if loc.startswith("condition")
                        else "RULE_PARSE_ERROR"
                    )
                    errors.append(
                        SemanticError(
                            code=code,
                            message=f"Rule '{name}' {loc or 'entry'}: {detail['msg']}",
                            path=f"{path}.{loc}" if loc else path,
                            span=span,
                        )
                    )
                continue
            rules[name] = rule

        # Pass two: references, levels, grain, cycles.
        for name, rule in list(rules.items()):
            path = f"rules.{name}"
            span = source_map.get(path) if source_map else None
            ok = True
            for field_name in condition_fields(rule.condition):
                if field_name not in known_fields:
                    ok = False
                    errors.append(
                        SemanticError(
                            code="UNKNOWN_RULE_FIELD",
                            message=(
                                f"Rule '{name}' compares '{field_name}', which is not a "
                                "dimension, measure or metric"
                            ),
                            path=f"{path}.condition",
                            span=span,
                            suggestions=_suggest_similar(field_name, sorted(known_fields)),
                        )
                    )
            for ref in condition_rule_refs(rule.condition):
                if ref == name or ref not in rules:
                    ok = False
                    errors.append(
                        SemanticError(
                            code="UNKNOWN_RULE",
                            message=f"Rule '{name}' references unknown rule '{ref}'"
                            if ref != name
                            else f"Rule '{name}' references itself",
                            path=f"{path}.condition",
                            span=span,
                            suggestions=_suggest_similar(ref, [r for r in rules if r != name]),
                        )
                    )
            for dim in rule.grain:
                if dim not in dimensions:
                    ok = False
                    errors.append(
                        SemanticError(
                            code="UNKNOWN_RULE_GRAIN",
                            message=f"Rule '{name}' grain names unknown dimension '{dim}'",
                            path=f"{path}.grain",
                            span=span,
                            suggestions=_suggest_similar(dim, list(dimensions)),
                        )
                    )
            if rule.severity is not None and rule.type in (
                RuleType.CLASSIFICATION,
                RuleType.ELIGIBILITY,
            ):
                ok = False
                errors.append(
                    SemanticError(
                        code="INVALID_RULE_SEVERITY",
                        message=(
                            f"Rule '{name}' is a {rule.type.value} rule; 'severity' belongs to "
                            "validation and constraint rules"
                        ),
                        path=f"{path}.severity",
                        span=span,
                    )
                )
            if not ok:
                continue
            level = rule_level(rule, rules, aggregate_names)
            if level == AGGREGATE:
                # The condition becomes HAVING over a GROUP BY of the grain, so a
                # dimension it compares must be a grain dimension: anything else
                # is neither grouped nor aggregated and strict engines reject it.
                outside = [
                    f
                    for f in transitive_fields(rule, rules)
                    if f in dimensions and f not in rule.grain
                ]
                for dim in outside:
                    errors.append(
                        SemanticError(
                            code="RULE_DIMENSION_OUTSIDE_GRAIN",
                            message=(
                                f"Rule '{name}' compares dimension '{dim}', which is not in its "
                                f"grain {rule.grain}; an aggregate rule may compare only its "
                                "grain dimensions, measures and metrics"
                            ),
                            path=f"{path}.condition",
                            span=span,
                        )
                    )
            if level == AGGREGATE and not rule.grain:
                errors.append(
                    SemanticError(
                        code="RULE_GRAIN_REQUIRED",
                        message=(
                            f"Rule '{name}' reads a measure or metric, so it needs a 'grain': "
                            "the dimensions it is evaluated at"
                        ),
                        path=path,
                        span=span,
                    )
                )
            elif level == ROW_LEVEL and rule.grain:
                errors.append(
                    SemanticError(
                        code="RULE_GRAIN_NOT_ALLOWED",
                        message=(
                            f"Rule '{name}' reads dimensions only; a 'grain' is for rules "
                            "over measures or metrics"
                        ),
                        path=f"{path}.grain",
                        span=span,
                    )
                )
            for ref in condition_rule_refs(rule.condition):
                other = rules[ref]
                other_level = rule_level(other, rules, aggregate_names)
                if other_level != level or (
                    level == AGGREGATE and set(other.grain) != set(rule.grain)
                ):
                    errors.append(
                        SemanticError(
                            code="RULE_REFERENCE_MISMATCH",
                            message=(
                                f"Rule '{name}' ({level}, grain {rule.grain}) references "
                                f"'{ref}' ({other_level}, grain {other.grain}); a referenced "
                                "rule must have the same level and grain"
                            ),
                            path=f"{path}.condition",
                            span=span,
                        )
                    )

        for cycle in self._rule_cycles(rules):
            first = cycle[0]
            errors.append(
                SemanticError(
                    code="CYCLIC_RULE_REFERENCE",
                    message="Rules reference each other in a cycle: " + " -> ".join(cycle),
                    path=f"rules.{first}.condition",
                    span=source_map.get(f"rules.{first}") if source_map else None,
                )
            )
        return rules

    def _check_condition_keys(self, raw: object, path: str, errors: list[SemanticError]) -> None:
        """UNKNOWN_PROPERTY for stray keys anywhere in a condition tree."""
        if not isinstance(raw, dict):
            return
        _check_unknown_keys(raw, _RULE_CONDITION_KEYS, path, errors)
        for key in ("all", "any"):
            children = raw.get(key)
            if isinstance(children, list):
                for i, child in enumerate(children):
                    self._check_condition_keys(child, f"{path}.{key}[{i}]", errors)
        self._check_condition_keys(raw.get("not"), f"{path}.not", errors)

    @staticmethod
    def _rule_cycles(rules: dict[str, Rule]) -> list[list[str]]:
        """Each reference cycle once, as the names along it plus the closing name."""
        graph: dict[str, list[str]] = {
            name: [r for r in condition_rule_refs(rule.condition) if r in rules and r != name]
            for name, rule in rules.items()
        }
        cycles: list[list[str]] = []
        reported: set[frozenset[str]] = set()
        state: dict[str, int] = {}
        stack: list[str] = []

        def visit(node: str) -> None:
            state[node] = 1
            stack.append(node)
            for nxt in graph[node]:
                if state.get(nxt, 0) == 0:
                    visit(nxt)
                elif state[nxt] == 1:
                    cycle = stack[stack.index(nxt) :] + [nxt]
                    key = frozenset(cycle)
                    if key not in reported:
                        reported.add(key)
                        cycles.append(cycle)
            stack.pop()
            state[node] = 2

        for name in rules:
            if state.get(name, 0) == 0:
                visit(name)
        return cycles

    def _parse_examples(self, raw: object, errors: list[SemanticError]) -> list[ModelExample]:
        """Parse the model-level ``examples:`` block.

        Accepts a list of mapping entries. Each entry must have ``name``,
        ``description``, and ``query``. ``intent_tags`` (alias ``intentTags``)
        is optional. Names must be unique within the block.
        """
        if raw is None:
            return []
        if not isinstance(raw, list):
            errors.append(
                SemanticError(
                    code="EXAMPLES_PARSE_ERROR",
                    message="'examples' must be a YAML list of example entries",
                    path="examples",
                )
            )
            return []

        out: list[ModelExample] = []
        seen: set[str] = set()
        for i, entry in enumerate(raw):
            if not isinstance(entry, dict):
                errors.append(
                    SemanticError(
                        code="EXAMPLES_PARSE_ERROR",
                        message=f"examples[{i}] must be a mapping",
                        path=f"examples[{i}]",
                    )
                )
                continue
            _check_unknown_keys(entry, _MODEL_EXAMPLE_KEYS, f"examples[{i}]", errors)
            name = entry.get("name")
            description = entry.get("description")
            query = entry.get("query")
            intent_tags = entry.get("intent_tags") or entry.get("intentTags") or []
            if not isinstance(name, str) or not name:
                errors.append(
                    SemanticError(
                        code="EXAMPLES_PARSE_ERROR",
                        message=f"examples[{i}].name is required and must be a string",
                        path=f"examples[{i}].name",
                    )
                )
                continue
            if name in seen:
                errors.append(
                    SemanticError(
                        code="DUPLICATE_EXAMPLE_NAME",
                        message=f"Duplicate example name '{name}'",
                        path=f"examples[{i}].name",
                    )
                )
                continue
            if not isinstance(description, str):
                errors.append(
                    SemanticError(
                        code="EXAMPLES_PARSE_ERROR",
                        message=f"examples[{i}].description is required",
                        path=f"examples[{i}].description",
                    )
                )
                continue
            if not isinstance(query, dict):
                errors.append(
                    SemanticError(
                        code="EXAMPLES_PARSE_ERROR",
                        message=f"examples[{i}].query must be a mapping (QueryObject payload)",
                        path=f"examples[{i}].query",
                    )
                )
                continue
            if not isinstance(intent_tags, list):
                errors.append(
                    SemanticError(
                        code="EXAMPLES_PARSE_ERROR",
                        message=f"examples[{i}].intent_tags must be a list",
                        path=f"examples[{i}].intent_tags",
                    )
                )
                continue
            seen.add(name)
            out.append(
                ModelExample(
                    name=name,
                    description=description,
                    intent_tags=[str(t) for t in intent_tags],
                    query=dict(query),
                )
            )
        return out

    def _validate_expression_refs(
        self,
        measure_name: str,
        expression: str,
        data_objects: dict[str, DataObject],
        errors: list[SemanticError],
        source_map: SourceMap | None,
    ) -> None:
        """Validate {[DataObject].[Column]} references in a measure expression."""
        span = source_map.get(f"measures.{measure_name}.expression") if source_map else None
        # The same scanner the tokenizer and the dependency walk use, so a
        # padded reference is read here exactly as it will be compiled.
        named_refs = find_qualified_refs(expression)
        for obj_name, col_name in named_refs:
            if obj_name not in data_objects:
                errors.append(
                    SemanticError(
                        code="UNKNOWN_DATA_OBJECT_IN_EXPRESSION",
                        message=(
                            f"Measure '{measure_name}' expression references unknown "
                            f"data object '{obj_name}'"
                        ),
                        path=f"measures.{measure_name}.expression",
                        span=span,
                    )
                )
            elif col_name not in data_objects[obj_name].columns:
                errors.append(
                    SemanticError(
                        code="UNKNOWN_COLUMN_IN_EXPRESSION",
                        message=(
                            f"Measure '{measure_name}' expression references unknown column "
                            f"'{col_name}' in data object '{obj_name}'"
                        ),
                        path=f"measures.{measure_name}.expression",
                        span=span,
                    )
                )

        # Strip valid refs, scan remainder for malformed attempts. The scan
        # itself lives in ``models.expressions``: the measure-expression parse
        # check consults it too, so that a botched bracket is reported once,
        # by the check that names the bracket.
        for ref, reason in find_malformed_measure_refs(expression):
            errors.append(
                SemanticError(
                    code="MALFORMED_EXPRESSION_REF",
                    message=(
                        f"Measure '{measure_name}' has malformed reference '{ref}' — {reason}"
                    ),
                    path=f"measures.{measure_name}.expression",
                    span=span,
                )
            )

    @staticmethod
    def _check_reaggregate_refs(
        name: str,
        raw_metric: dict[str, Any],
        measures: dict[str, Measure],
        synthesized_measures: set[str],
        raw_metrics: dict[str, Any],
        dimensions: dict[str, Dimension],
        errors: list[SemanticError],
        source_map: SourceMap | None,
    ) -> None:
        """Check what a reaggregate metric's ``measure``, ``having`` and ``per`` name.

        The first stage is ``measure`` as a query of its own at a finer grain:
        a measure, or another reaggregate metric, whose two stages then run
        inside it. Any other metric is refused, and so is a chain of reaggregate
        metrics that comes back to one it started from. A ``having`` condition
        is that query's HAVING and names a measure: a metric there could be the
        one it filters, or reach it again.
        """
        path = f"metrics.{name}"

        def report(
            code: str, field: str, message: str, suggestions: list[str] | None = None
        ) -> None:
            errors.append(
                SemanticError(
                    code=code,
                    message=message,
                    path=f"{path}.{field}",
                    span=source_map.get(f"{path}.{field}") if source_map else None,
                    suggestions=suggestions or [],
                )
            )

        ref = raw_metric.get("measure")
        if isinstance(ref, str) and ref:
            measure = measures.get(ref)
            if ref in raw_metrics and measure is None:
                if _is_cumulative(raw_metrics[ref]):
                    # Stage 1 plans the cumulative as a query of its own at the
                    # query's dimensions plus ``per``; without its time
                    # dimension there it would be evaluated as of one period.
                    time_dim = raw_metrics[ref].get("timeDimension")
                    per_names = raw_metric.get("per") or []
                    if time_dim and time_dim not in per_names:
                        report(
                            "REAGGREGATE_CUMULATIVE_TIME_DIMENSION",
                            "per",
                            f"Reaggregate metric '{name}' reaggregates the cumulative metric "
                            f"'{ref}', so 'per' has to name its timeDimension '{time_dim}': "
                            f"the first stage then holds one running value per period.",
                        )
                    if raw_metric.get("population") == ReaggregatePopulation.DIMENSION:
                        report(
                            "REAGGREGATE_POPULATION_PER",
                            "population",
                            f"Reaggregate metric '{name}' reaggregates the cumulative metric "
                            f"'{ref}' per its time dimension, a period with no data object "
                            f"of members to list; its population is the facts'.",
                        )
                    if cycle := _reaggregate_cycle(name, raw_metrics):
                        report(
                            "REAGGREGATE_CYCLE",
                            "measure",
                            "Reaggregate metrics reference each other in a cycle: "
                            + " -> ".join(cycle),
                        )
                elif not _is_reaggregate(raw_metrics[ref]):
                    report(
                        "REAGGREGATE_MEASURE_ONLY",
                        "measure",
                        f"Reaggregate metric '{name}' references metric '{ref}'; "
                        f"'measure' must name a measure, a reaggregate metric or a "
                        f"cumulative metric.",
                    )
                elif cycle := _reaggregate_cycle(name, raw_metrics):
                    report(
                        "REAGGREGATE_CYCLE",
                        "measure",
                        "Reaggregate metrics reference each other in a cycle: "
                        + " -> ".join(cycle),
                    )
            elif measure is None and ref not in synthesized_measures:
                report(
                    "UNKNOWN_MEASURE",
                    "measure",
                    f"Reaggregate metric '{name}' references unknown measure '{ref}'",
                    _suggest_similar(ref, [*measures, *synthesized_measures]),
                )
            if raw_metric.get("aggregation") == ReaggregateAggType.AVG and (
                averages := _average_path(ref, measures, raw_metrics)
            ):
                source = (
                    "which is itself an average"
                    if len(averages) == 1
                    else f"which takes its value from the average '{averages[-1]}' ("
                    + " -> ".join(averages)
                    + ")"
                )
                report(
                    "REAGGREGATE_AVG_OF_AVG",
                    "aggregation",
                    f"Reaggregate metric '{name}' averages '{ref}', {source}. An average "
                    f"of averages weighs every group alike, however many rows it averages; "
                    f"use another aggregation, or average the underlying measure per the "
                    f"combined 'per' dimensions.",
                )

        having = raw_metric.get("having")
        for item in having if isinstance(having, list) else []:
            field = item.get("field") if isinstance(item, dict) else None
            if not isinstance(field, str) or field in measures or field in synthesized_measures:
                continue
            if field in raw_metrics:
                report(
                    "REAGGREGATE_HAVING_MEASURE_ONLY",
                    "having",
                    f"Reaggregate metric '{name}' has a 'having' condition on metric "
                    f"'{field}'; a condition on the first stage must name a measure.",
                )
            else:
                report(
                    "UNKNOWN_MEASURE",
                    "having",
                    f"Reaggregate metric '{name}' has a 'having' condition on unknown "
                    f"measure '{field}'",
                    _suggest_similar(field, [*measures, *synthesized_measures]),
                )

        if raw_metric.get("carryForward") is True and (
            raw_metric.get("population") != ReaggregatePopulation.DIMENSION
        ):
            # Carrying needs the members' grid: with the facts' population a
            # member with no row in a period has no row to carry into.
            report(
                "REAGGREGATE_CARRY_FORWARD_POPULATION",
                "carryForward",
                f"Reaggregate metric '{name}' carries a member's value forward, which "
                f"needs every member in every period: set 'population: dimension'.",
            )
        per = raw_metric.get("per")
        if not isinstance(per, list):
            return
        if raw_metric.get("population") == ReaggregatePopulation.DIMENSION:
            # The members are read from the dimensions' own data object: a time
            # bucket has no member table, a role reached through a fact (``via``)
            # is defined by that fact, and a ``having`` would bring a member it
            # removes back with the default value.
            for entry in per:
                if not isinstance(entry, str):
                    continue
                declared = dimensions.get(entry)
                # An unknown name is REAGGREGATE_UNKNOWN_DIMENSION's. A declared
                # ``timeGrain`` makes the entry a time bucket as ``:grain`` does.
                if ":" in entry or (
                    declared is not None
                    and (declared.via is not None or declared.time_grain is not None)
                ):
                    report(
                        "REAGGREGATE_POPULATION_PER",
                        "per",
                        f"Reaggregate metric '{name}' takes its population from the "
                        f"dimension, so each 'per' entry has to be a dimension of its own "
                        f"data object, without a time grain (written or declared) or "
                        f"'via'; '{entry}' is not.",
                    )
            if raw_metric.get("having"):
                report(
                    "REAGGREGATE_POPULATION_HAVING",
                    "having",
                    f"Reaggregate metric '{name}' takes its population from the dimension, "
                    f"where 'having' would bring a member it removes back with the default "
                    f"value; filter the members with the query instead.",
                )
        if raw_metric.get("aggregation") in (ReaggregateAggType.FIRST, ReaggregateAggType.LAST):
            entries = [e for e in per if isinstance(e, str)] if isinstance(per, list) else []
            times = time_per_entries(entries, dimensions)
            if len(times) != 1:
                found = f"found {', '.join(repr(t) for t in times)}" if times else "found none"
                report(
                    "REAGGREGATE_ORDER_KEY",
                    "per",
                    f"Reaggregate metric '{name}' takes the {raw_metric['aggregation']} group "
                    "by time, so 'per' needs exactly one time dimension to order by (a grain "
                    f"such as 'Order Date:month', or a date or timestamp dimension); {found}.",
                )
        for entry in per:
            if not isinstance(entry, str):
                continue
            try:
                dim_ref = DimensionRef.parse(entry)
            except ValueError:
                report(
                    "REAGGREGATE_INVALID_PER",
                    "per",
                    f"Reaggregate metric '{name}' has 'per' entry '{entry}' with an unknown "
                    f"time grain; use one of {', '.join(g.value for g in TimeGrain)}.",
                )
                continue
            if dim_ref.name not in dimensions:
                report(
                    "REAGGREGATE_UNKNOWN_DIMENSION",
                    "per",
                    f"Reaggregate metric '{name}' references unknown dimension '{dim_ref.name}'",
                    _suggest_similar(dim_ref.name, list(dimensions)),
                )

    def _validate_metric_expression_refs(
        self,
        metric_name: str,
        expression: str,
        measures: dict[str, Measure],
        errors: list[SemanticError],
        source_map: SourceMap | None,
        metrics: dict[str, Metric] | None = None,
        synthesized_measures: set[str] | None = None,
        composable_metric_types: tuple[MetricType, ...] = (),
    ) -> None:
        """Validate {[Measure Name]} references in a metric expression.

        References resolve to measures. ``synthesized_measures`` names the
        auto-generated ``<object> Count`` measures, which are valid references
        even though they are not in ``measures``.

        A reference to another *metric* is only valid for the compositions the
        compiler actually expands, named by *composable_metric_types*: a derived
        metric over a **window** metric, which the window wrapper substitutes as
        a column of its base CTE. Every other metric-over-metric reference is
        refused here rather than compiled: the planner substitutes a metric's
        components one level only, so the inner metric's own placeholders
        survive into the SQL as bare column names that no engine can bind
        (``Referenced column "Revenue" not found in FROM clause``).

        ``metrics`` defaults to ``None`` so existing callers continue to work;
        the caller passes the in-progress metrics dict so a reference can be
        classified.
        """
        span = source_map.get(f"metrics.{metric_name}.expression") if source_map else None

        valid_refs = re.findall(r"\{\[([^\]{}\[]+)\]\}", expression)

        # Strip valid {[Name]} refs, then scan remainder for malformed attempts.
        remainder = re.sub(r"\{\[[^\]{}\[]+\]\}", "", expression)

        # {[Name} — missing closing ]
        for bad in re.findall(r"\{\[([^\]{}]*)\}", remainder):
            errors.append(
                SemanticError(
                    code="MALFORMED_EXPRESSION_REF",
                    message=(
                        f"Metric '{metric_name}' has malformed reference"
                        f" '{{[{bad}}}' — missing closing ']'"
                    ),
                    path=f"metrics.{metric_name}.expression",
                    span=span,
                )
            )

        # {[Name] — missing closing }
        for bad in re.findall(r"\{\[([^\]{}]+)\](?!\})", remainder):
            errors.append(
                SemanticError(
                    code="MALFORMED_EXPRESSION_REF",
                    message=(
                        f"Metric '{metric_name}' has malformed reference"
                        f" '{{[{bad}]' — missing closing '}}'"
                    ),
                    path=f"metrics.{metric_name}.expression",
                    span=span,
                )
            )

        # {Name]} — missing opening [
        for bad in re.findall(r"\{([^\[{}\]]+)\]\}", remainder):
            errors.append(
                SemanticError(
                    code="MALFORMED_EXPRESSION_REF",
                    message=(
                        f"Metric '{metric_name}' has malformed reference"
                        f" '{{{bad}]}}' — missing opening '['"
                    ),
                    path=f"metrics.{metric_name}.expression",
                    span=span,
                )
            )

        # {Name} — missing both [ and ]
        for bad in re.findall(r"\{([^\[{\]}\s]+)\}", remainder):
            errors.append(
                SemanticError(
                    code="MALFORMED_EXPRESSION_REF",
                    message=(
                        f"Metric '{metric_name}' has malformed reference"
                        f" '{{{bad}}}' — missing '[' and ']'"
                    ),
                    path=f"metrics.{metric_name}.expression",
                    span=span,
                )
            )

        # [Name]} — missing opening {
        for bad in re.findall(r"(?<!\{)\[([^\]{}\[]+)\]\}", remainder):
            errors.append(
                SemanticError(
                    code="MALFORMED_EXPRESSION_REF",
                    message=(
                        f"Metric '{metric_name}' has malformed reference"
                        f" '[{bad}]}}' — missing opening '{{'"
                    ),
                    path=f"metrics.{metric_name}.expression",
                    span=span,
                )
            )

        known_metrics = metrics or {}
        known_counts = synthesized_measures or set()
        for ref_name in valid_refs:
            if ref_name in measures or ref_name in known_counts:
                continue

            referenced = known_metrics.get(ref_name)
            if referenced is None:
                errors.append(
                    SemanticError(
                        code="UNKNOWN_MEASURE_REF",
                        message=(f"Metric '{metric_name}' references unknown measure '{ref_name}'"),
                        path=f"metrics.{metric_name}.expression",
                        span=span,
                        suggestions=_suggest_similar(
                            ref_name,
                            list(measures.keys())
                            + list(known_metrics.keys())
                            + sorted(known_counts),
                        ),
                    )
                )
                continue

            if referenced.type not in composable_metric_types:
                allowed = ", ".join(sorted(t.value for t in composable_metric_types))
                permitted = f" or a {allowed} metric" if allowed else ""
                errors.append(
                    SemanticError(
                        code="UNSUPPORTED_METRIC_REF",
                        message=(
                            f"Metric '{metric_name}' references metric '{ref_name}', which is a "
                            f"{referenced.type.value} metric. A metric expression can reference "
                            f"measures{permitted} — nesting one metric inside another is not "
                            f"supported, and the query would compile to a column reference no "
                            f"engine can resolve."
                        ),
                        path=f"metrics.{metric_name}.expression",
                        span=span,
                        hint=(
                            f"Reference the measures '{ref_name}' is built from, or inline its "
                            f"expression into '{metric_name}'."
                        ),
                        context={"metric": metric_name, "references": ref_name},
                    )
                )

resolve(raw, source_map=None)

Resolve raw YAML dict into a validated SemanticModel.

Returns (model, validation_result). If there are errors, the model may be partially populated.

Source code in src/orionbelt/parser/resolver.py

def resolve(
    self,
    raw: dict[str, Any],
    source_map: SourceMap | None = None,
) -> tuple[SemanticModel, ValidationResult]:
    """Resolve raw YAML dict into a validated SemanticModel.

    Returns (model, validation_result). If there are errors,
    the model may be partially populated.
    """
    errors: list[SemanticError] = []
    warnings: list[SemanticError] = []

    # Strict OBML: reject unknown top-level keys (catches typos like
    # ``dataObjekt:`` that would silently be dropped by ``raw.get(...)``).
    _check_unknown_keys(raw, _TOP_LEVEL_KEYS, "", errors, source_map)

    # Ontology prefixes come first: every object's externalConceptMappings
    # expand against them.
    ontology = _parse_ontology(raw.get("ontology"), errors, source_map)
    prefixes = effective_prefixes(ontology.prefixes if ontology else None)

    # Parse data objects
    data_objects: dict[str, DataObject] = {}
    raw_objects = raw.get("dataObjects", {})
    if not isinstance(raw_objects, dict):
        errors.append(
            SemanticError(
                code="DATA_OBJECT_PARSE_ERROR",
                message="'dataObjects' must be a YAML mapping, not a list or scalar",
                path="dataObjects",
            )
        )
        raw_objects = {}
    for name, raw_obj in raw_objects.items():
        try:
            _check_unknown_keys(
                raw_obj, _DATA_OBJECT_KEYS, f"dataObjects.{name}", errors, source_map
            )
            obj_columns: dict[str, DataObjectColumn] = {}
            for fname, fdata in raw_obj.get("columns", {}).items():
                _check_unknown_keys(
                    fdata,
                    _DATA_OBJECT_COLUMN_KEYS,
                    f"dataObjects.{name}.columns.{fname}",
                    errors,
                    source_map,
                )
                obj_columns[fname] = DataObjectColumn(
                    name=fname,
                    code=fdata.get("code", fname if not fdata.get("expression") else ""),
                    abstract_type=fdata.get("abstractType", "string"),
                    sql_type=fdata.get("sqlType"),
                    sql_precision=fdata.get("sqlPrecision"),
                    sql_scale=fdata.get("sqlScale"),
                    num_class=fdata.get("numClass"),
                    primary_key=bool(fdata.get("primaryKey", False)),
                    description=fdata.get("description"),
                    comment=fdata.get("comment"),
                    owner=fdata.get("owner"),
                    expression=fdata.get("expression"),
                    synonyms=fdata.get("synonyms", []),
                    custom_extensions=_parse_extensions(fdata),
                )

            obj_joins: list[DataObjectJoin] = []
            for ji, jdata in enumerate(raw_obj.get("joins", [])):
                _check_unknown_keys(
                    jdata,
                    _DATA_OBJECT_JOIN_KEYS,
                    f"dataObjects.{name}.joins[{ji}]",
                    errors,
                    source_map,
                )
                obj_joins.append(
                    DataObjectJoin(
                        join_type=jdata["joinType"],
                        join_to=jdata["joinTo"],
                        columns_from=jdata["columnsFrom"],
                        columns_to=jdata["columnsTo"],
                        secondary=jdata.get("secondary", False),
                        path_name=jdata.get("pathName"),
                        required=jdata.get("required", False),
                    )
                )

            data_objects[name] = DataObject(
                name=name,
                code=raw_obj.get("code", ""),
                database=raw_obj.get("database", ""),
                schema_name=raw_obj.get("schema", ""),
                columns=obj_columns,
                joins=obj_joins,
                description=raw_obj.get("description"),
                comment=raw_obj.get("comment"),
                owner=raw_obj.get("owner"),
                countable=raw_obj.get("countable", True),
                count_label=raw_obj.get("countLabel"),
                synonyms=raw_obj.get("synonyms", []),
                custom_extensions=_parse_extensions(raw_obj),
                external_concept_mappings=_parse_concept_mappings(
                    raw_obj, f"dataObjects.{name}", prefixes, errors, source_map
                ),
                refresh=_parse_refresh(raw_obj.get("refresh"), name, errors),
                nested_in=_parse_nested_in(raw_obj.get("nestedIn")),
            )
        except Exception as e:
            span = source_map.get(f"dataObjects.{name}") if source_map else None
            errors.append(
                SemanticError(
                    code="DATA_OBJECT_PARSE_ERROR",
                    message=f"Failed to parse data object '{name}': {e}",
                    path=f"dataObjects.{name}",
                    span=span,
                )
            )

    # Parse dimensions
    dimensions: dict[str, Dimension] = {}
    raw_dims = raw.get("dimensions", {})
    if not isinstance(raw_dims, dict):
        errors.append(
            SemanticError(
                code="DIMENSION_PARSE_ERROR",
                message="'dimensions' must be a YAML mapping, not a list or scalar",
                path="dimensions",
            )
        )
        raw_dims = {}
    for name, raw_dim in raw_dims.items():
        try:
            _check_unknown_keys(
                raw_dim, _DIMENSION_KEYS, f"dimensions.{name}", errors, source_map
            )
            data_object = raw_dim.get("dataObject")
            column = raw_dim.get("column")

            # Validate the data object exists
            if data_object and data_object not in data_objects:
                span = source_map.get(f"dimensions.{name}") if source_map else None
                errors.append(
                    SemanticError(
                        code="UNKNOWN_DATA_OBJECT",
                        message=(
                            f"Dimension '{name}' references unknown data object '{data_object}'"
                        ),
                        path=f"dimensions.{name}",
                        span=span,
                        suggestions=_suggest_similar(data_object, list(data_objects.keys())),
                    )
                )

            # Validate the column exists in the data object
            if (
                data_object
                and column
                and data_object in data_objects
                and column not in data_objects[data_object].columns
            ):
                span = source_map.get(f"dimensions.{name}") if source_map else None
                errors.append(
                    SemanticError(
                        code="UNKNOWN_COLUMN",
                        message=(
                            f"Dimension '{name}' references unknown column "
                            f"'{column}' in data object '{data_object}'"
                        ),
                        path=f"dimensions.{name}",
                        span=span,
                        suggestions=_suggest_similar(
                            column, list(data_objects[data_object].columns.keys())
                        ),
                    )
                )

            via = raw_dim.get("via")
            if via and via not in data_objects:
                span = source_map.get(f"dimensions.{name}") if source_map else None
                errors.append(
                    SemanticError(
                        code="UNKNOWN_DATA_OBJECT",
                        message=(
                            f"Dimension '{name}' via references unknown data object '{via}'"
                        ),
                        path=f"dimensions.{name}",
                        span=span,
                        suggestions=_suggest_similar(via, list(data_objects.keys())),
                    )
                )

            dimensions[name] = Dimension(
                name=name,
                view=data_object or "",
                column=column or "",
                result_type=raw_dim.get("resultType", "string"),
                time_grain=raw_dim.get("timeGrain"),
                via=via,
                path_name=raw_dim.get("pathName"),
                description=raw_dim.get("description"),
                format=raw_dim.get("format"),
                owner=raw_dim.get("owner"),
                synonyms=raw_dim.get("synonyms", []),
                custom_extensions=_parse_extensions(raw_dim),
                external_concept_mappings=_parse_concept_mappings(
                    raw_dim, f"dimensions.{name}", prefixes, errors, source_map
                ),
            )
        except Exception as e:
            span = source_map.get(f"dimensions.{name}") if source_map else None
            errors.append(
                SemanticError(
                    code="DIMENSION_PARSE_ERROR",
                    message=f"Failed to parse dimension '{name}': {e}",
                    path=f"dimensions.{name}",
                    span=span,
                )
            )

    # Parse measures
    measures: dict[str, Measure] = {}
    raw_measures = raw.get("measures", {})
    if not isinstance(raw_measures, dict):
        errors.append(
            SemanticError(
                code="MEASURE_PARSE_ERROR",
                message="'measures' must be a YAML mapping, not a list or scalar",
                path="measures",
            )
        )
        raw_measures = {}
    for name, raw_meas in raw_measures.items():
        try:
            _check_unknown_keys(raw_meas, _MEASURE_KEYS, f"measures.{name}", errors, source_map)
            measure_columns: list[DataColumnRef] = []
            for ci, fdata in enumerate(raw_meas.get("columns", [])):
                _check_unknown_keys(
                    fdata,
                    _DATA_COLUMN_REF_KEYS,
                    f"measures.{name}.columns[{ci}]",
                    errors,
                    source_map,
                )
                measure_columns.append(
                    DataColumnRef(
                        view=fdata.get("dataObject"),
                        column=fdata.get("column"),
                    )
                )

            # Resolve expression field references
            expression = raw_meas.get("expression")
            if expression:
                self._validate_expression_refs(
                    name, expression, data_objects, errors, source_map
                )

            # Parse measure filters (new `filters:` list or legacy `filter:` single)
            measure_filters: list[MeasureFilterItem] = []
            raw_filters = raw_meas.get("filters")
            if raw_filters and isinstance(raw_filters, list):
                for fi, rf in enumerate(raw_filters):
                    measure_filters.append(
                        _parse_measure_filter_item(
                            rf,
                            f"measures.{name}.filters[{fi}]",
                            errors,
                            source_map,
                        )
                    )
            else:
                # Backward compat: single `filter:` key → [filter]
                raw_filter = raw_meas.get("filter")
                if raw_filter:
                    measure_filters.append(
                        _parse_measure_filter_item(
                            raw_filter, f"measures.{name}.filter", errors, source_map
                        )
                    )

            # Parse grain override
            grain_override: GrainOverride | None = None
            raw_grain = raw_meas.get("grain")
            if raw_grain and isinstance(raw_grain, dict):
                _check_unknown_keys(
                    raw_grain,
                    _GRAIN_OVERRIDE_KEYS,
                    f"measures.{name}.grain",
                    errors,
                    source_map,
                )
                grain_override = GrainOverride(
                    mode=raw_grain.get("mode", "RELATIVE"),
                    exclude=raw_grain.get("exclude", []),
                    include=raw_grain.get("include", []),
                    keep_only=raw_grain.get("keepOnly", []),
                )
                # Validate dimension references in grain
                for dim_name in (
                    grain_override.include + grain_override.exclude + grain_override.keep_only
                ):
                    if dim_name not in dimensions:
                        span = source_map.get(f"measures.{name}.grain") if source_map else None
                        errors.append(
                            SemanticError(
                                code="UNKNOWN_GRAIN_DIMENSION",
                                message=(
                                    f"Measure '{name}' grain references "
                                    f"unknown dimension '{dim_name}'"
                                ),
                                path=f"measures.{name}.grain",
                                span=span,
                                suggestions=_suggest_similar(dim_name, list(dimensions.keys())),
                            )
                        )

            # Parse filter context
            filter_ctx: FilterContext | None = None
            raw_fc = raw_meas.get("filterContext")
            if raw_fc and isinstance(raw_fc, dict):
                _check_unknown_keys(
                    raw_fc,
                    _FILTER_CONTEXT_KEYS,
                    f"measures.{name}.filterContext",
                    errors,
                    source_map,
                )
                include_filters: list[FilterContextFilter] = []
                for inc_i, raw_incl in enumerate(raw_fc.get("include", [])):
                    if isinstance(raw_incl, dict):
                        _check_unknown_keys(
                            raw_incl,
                            _FILTER_CONTEXT_FILTER_KEYS,
                            f"measures.{name}.filterContext.include[{inc_i}]",
                            errors,
                            source_map,
                        )
                        include_filters.append(
                            FilterContextFilter(
                                field=raw_incl.get("field", ""),
                                op=raw_incl.get("op", "equals"),
                                value=raw_incl.get("value"),
                            )
                        )
                filter_ctx = FilterContext(
                    mode=raw_fc.get("mode", "RELATIVE"),
                    exclude=raw_fc.get("exclude", []),
                    include=include_filters,
                    keep_only=raw_fc.get("keepOnly", []),
                )
                # Validate field references in exclude/keepOnly
                all_dim_names = set(dimensions.keys())
                all_col_refs: set[str] = set()
                for obj_name, obj_def in data_objects.items():
                    for col_name in obj_def.columns:
                        all_col_refs.add(f"{obj_name}.{col_name}")
                for field_name in filter_ctx.exclude + filter_ctx.keep_only:
                    if field_name not in all_dim_names and field_name not in all_col_refs:
                        span = (
                            source_map.get(f"measures.{name}.filterContext")
                            if source_map
                            else None
                        )
                        errors.append(
                            SemanticError(
                                code="UNKNOWN_FILTER_CONTEXT_FIELD",
                                message=(
                                    f"Measure '{name}' filterContext references "
                                    f"unknown field '{field_name}'"
                                ),
                                path=f"measures.{name}.filterContext",
                                span=span,
                                suggestions=_suggest_similar(field_name, list(all_dim_names)),
                            )
                        )
                for incl in filter_ctx.include:
                    if incl.field not in all_dim_names and incl.field not in all_col_refs:
                        span = (
                            source_map.get(f"measures.{name}.filterContext")
                            if source_map
                            else None
                        )
                        errors.append(
                            SemanticError(
                                code="UNKNOWN_FILTER_CONTEXT_FIELD",
                                message=(
                                    f"Measure '{name}' filterContext.include "
                                    f"references unknown field '{incl.field}'"
                                ),
                                path=f"measures.{name}.filterContext.include",
                                span=span,
                                suggestions=_suggest_similar(incl.field, list(all_dim_names)),
                            )
                        )

            measures[name] = Measure(
                name=name,
                columns=measure_columns,
                result_type=raw_meas.get("resultType", "float"),
                aggregation=raw_meas.get("aggregation", "sum"),
                expression=expression,
                distinct=raw_meas.get("distinct", False),
                total=raw_meas.get("total", False),
                default_value=raw_meas.get("defaultValue"),
                anchor=raw_meas.get("anchor"),
                grain=grain_override,
                filter_context=filter_ctx,
                filters=measure_filters,
                data_type=raw_meas.get("dataType"),
                description=raw_meas.get("description"),
                format=raw_meas.get("format"),
                allow_fan_out=raw_meas.get("allowFanOut", False),
                delimiter=raw_meas.get("delimiter"),
                percentile=raw_meas.get("percentile"),
                within_group=raw_meas.get("withinGroup"),
                owner=raw_meas.get("owner"),
                synonyms=raw_meas.get("synonyms", []),
                custom_extensions=_parse_extensions(raw_meas),
                external_concept_mappings=_parse_concept_mappings(
                    raw_meas, f"measures.{name}", prefixes, errors, source_map
                ),
            )
        except Exception as e:
            span = source_map.get(f"measures.{name}") if source_map else None
            errors.append(
                SemanticError(
                    code="MEASURE_PARSE_ERROR",
                    message=f"Failed to parse measure '{name}': {e}",
                    path=f"measures.{name}",
                    span=span,
                )
            )

    # Validate the count-synthesis knobs here so a bad value becomes a
    # structured SemanticError rather than a raw AttributeError (list
    # pattern) or an uncaught Pydantic ValidationError (invalid token) at
    # model construction below. Fall back to safe values so resolution can
    # continue collecting errors.
    _count_pattern = raw.get("countLabelPattern", DEFAULT_COUNT_PATTERN)
    _pattern_err = count_pattern_error(_count_pattern)
    if _pattern_err is not None:
        span = source_map.get("countLabelPattern") if source_map else None
        errors.append(
            SemanticError(
                code="INVALID_COUNT_LABEL_PATTERN",
                message=_pattern_err,
                path="countLabelPattern",
                span=span,
            )
        )
        _count_pattern = DEFAULT_COUNT_PATTERN
    _expose_counts = raw.get("exposeCounts", True)
    if not isinstance(_expose_counts, bool):
        span = source_map.get("exposeCounts") if source_map else None
        errors.append(
            SemanticError(
                code="INVALID_EXPOSE_COUNTS",
                message="exposeCounts must be a boolean (true/false)",
                path="exposeCounts",
                span=span,
            )
        )
        _expose_counts = True

    # Names of synthesized count measures (name == resolved count label,
    # e.g. "Sales Count"). These are valid measure references (metrics may
    # target them) even though they are not declared — they are materialized
    # on read via ``effective_measures`` (see models/synthesis.py). Declared
    # measures already sit in ``measures``; a declared count of the same
    # name overrides synthesis, so unioning is safe either way.
    synthesized_measure_names: set[str] = (
        {
            count_label(key, obj, _count_pattern)
            for key, obj in data_objects.items()
            if obj.countable
        }
        if _expose_counts
        else set()
    )

    # Parse metrics
    metrics: dict[str, Metric] = {}
    raw_metrics = raw.get("metrics", {})
    if not isinstance(raw_metrics, dict):
        errors.append(
            SemanticError(
                code="METRIC_PARSE_ERROR",
                message="'metrics' must be a YAML mapping, not a list or scalar",
                path="metrics",
            )
        )
        raw_metrics = {}
    for name, raw_metric in raw_metrics.items():
        try:
            _check_unknown_keys(raw_metric, _METRIC_KEYS, f"metrics.{name}", errors, source_map)
            raw_pop_block = raw_metric.get("periodOverPeriod")
            if isinstance(raw_pop_block, dict):
                _check_unknown_keys(
                    raw_pop_block,
                    _PERIOD_OVER_PERIOD_KEYS,
                    f"metrics.{name}.periodOverPeriod",
                    errors,
                    source_map,
                )
            metric_type = raw_metric.get("type", "derived")
            # The other branches pass only their own fields to the model, so
            # these would be dropped silently rather than refused there.
            if metric_type != MetricType.REAGGREGATE and any(
                key in raw_metric
                for key in (
                    "per",
                    "aggregation",
                    "having",
                    "percentile",
                    "population",
                    "carryForward",
                )
            ):
                raise ValueError(
                    "'per', 'aggregation', 'having', 'percentile', 'population' and "
                    "'carryForward' are only valid on reaggregate metrics"
                )
            concept_links = _parse_concept_mappings(
                raw_metric, f"metrics.{name}", prefixes, errors, source_map
            )

            if metric_type == MetricType.CUMULATIVE:
                # Cumulative metric: validate measure reference exists
                ref_measure = raw_metric.get("measure", "")
                if (
                    ref_measure
                    and ref_measure not in measures
                    and ref_measure not in synthesized_measure_names
                    and not _is_reaggregate(raw_metrics.get(ref_measure))
                ):
                    span = source_map.get(f"metrics.{name}.measure") if source_map else None
                    errors.append(
                        SemanticError(
                            code="UNKNOWN_MEASURE",
                            message=(
                                f"Cumulative metric '{name}' references "
                                f"unknown measure '{ref_measure}'"
                            ),
                            path=f"metrics.{name}.measure",
                            span=span,
                        )
                    )

                # Validate timeDimension references a known dimension
                cum_time_dim = raw_metric.get("timeDimension", "")
                if cum_time_dim and cum_time_dim not in dimensions:
                    span = (
                        source_map.get(f"metrics.{name}.timeDimension") if source_map else None
                    )
                    errors.append(
                        SemanticError(
                            code="CUMULATIVE_UNKNOWN_TIME_DIMENSION",
                            message=(
                                f"Cumulative metric '{name}' references "
                                f"unknown time dimension '{cum_time_dim}'"
                            ),
                            path=f"metrics.{name}.timeDimension",
                            span=span,
                            suggestions=_suggest_similar(cum_time_dim, list(dimensions.keys())),
                        )
                    )

                metrics[name] = Metric(
                    name=name,
                    type=MetricType.CUMULATIVE,
                    measure=raw_metric.get("measure"),
                    time_dimension=raw_metric.get("timeDimension"),
                    cumulative_type=raw_metric.get("cumulativeType", "sum"),
                    window=raw_metric.get("window"),
                    grain_to_date=raw_metric.get("grainToDate"),
                    partition_by=list(raw_metric.get("partitionBy", []) or []),
                    data_type=raw_metric.get("dataType"),
                    description=raw_metric.get("description"),
                    format=raw_metric.get("format"),
                    owner=raw_metric.get("owner"),
                    synonyms=raw_metric.get("synonyms", []),
                    custom_extensions=_parse_extensions(raw_metric),
                    external_concept_mappings=concept_links,
                )
            elif metric_type == MetricType.PERIOD_OVER_PERIOD:
                # Period-over-period metric: validate expression + PoP config.
                # Its base has to be re-aggregated per period from the fact
                # tables, so the reference must be a measure.
                expression = raw_metric.get("expression", "")
                self._validate_metric_expression_refs(
                    name,
                    expression,
                    measures,
                    errors,
                    source_map,
                    metrics,
                    synthesized_measure_names,
                )

                raw_pop = raw_metric.get("periodOverPeriod")
                if not raw_pop:
                    span = source_map.get(f"metrics.{name}") if source_map else None
                    errors.append(
                        SemanticError(
                            code="METRIC_PARSE_ERROR",
                            message=(
                                f"Period-over-period metric '{name}' "
                                f"requires 'periodOverPeriod' configuration"
                            ),
                            path=f"metrics.{name}",
                            span=span,
                        )
                    )
                    raw_pop = {}

                # Validate time dimension reference
                pop_time_dim = raw_pop.get("timeDimension", "")
                if pop_time_dim and pop_time_dim not in dimensions:
                    span = (
                        source_map.get(f"metrics.{name}.periodOverPeriod")
                        if source_map
                        else None
                    )
                    errors.append(
                        SemanticError(
                            code="POP_UNKNOWN_TIME_DIMENSION",
                            message=(
                                f"Period-over-period metric '{name}' references "
                                f"unknown time dimension '{pop_time_dim}'"
                            ),
                            path=f"metrics.{name}.periodOverPeriod.timeDimension",
                            span=span,
                            suggestions=_suggest_similar(pop_time_dim, list(dimensions.keys())),
                        )
                    )

                pop_config = PeriodOverPeriod(
                    time_dimension=raw_pop.get("timeDimension", ""),
                    grain=raw_pop.get("grain", "month"),
                    offset=raw_pop.get("offset", -1),
                    offset_grain=raw_pop.get("offsetGrain", "year"),
                    comparison=raw_pop.get("comparison", "percentChange"),
                )

                metrics[name] = Metric(
                    name=name,
                    type=MetricType.PERIOD_OVER_PERIOD,
                    expression=expression,
                    period_over_period=pop_config,
                    data_type=raw_metric.get("dataType"),
                    description=raw_metric.get("description"),
                    format=raw_metric.get("format"),
                    owner=raw_metric.get("owner"),
                    synonyms=raw_metric.get("synonyms", []),
                    custom_extensions=_parse_extensions(raw_metric),
                    external_concept_mappings=concept_links,
                )
            elif metric_type == MetricType.REAGGREGATE:
                self._check_reaggregate_refs(
                    name,
                    raw_metric,
                    measures,
                    synthesized_measure_names,
                    raw_metrics,
                    dimensions,
                    errors,
                    source_map,
                )
                # Every recognised key is passed on, so a field of another
                # metric type is refused by the model instead of dropped.
                metrics[name] = Metric.model_validate(
                    {
                        **{
                            key: value
                            for key, value in raw_metric.items()
                            if key in _METRIC_KEYS
                            and key not in ("customExtensions", "externalConceptMappings")
                        },
                        "name": name,
                        "customExtensions": _parse_extensions(raw_metric),
                        "externalConceptMappings": concept_links,
                    }
                )
            elif metric_type == MetricType.WINDOW:
                # Window metric (rank/lag/lead/ntile/first_value/last_value)
                ref_measure = raw_metric.get("measure")
                # A reaggregate metric is a base too: the window then runs
                # over its finished values (``compiler/late_window.py``).
                if (
                    ref_measure
                    and ref_measure not in measures
                    and ref_measure not in synthesized_measure_names
                    and not _is_reaggregate(raw_metrics.get(ref_measure))
                ):
                    span = source_map.get(f"metrics.{name}.measure") if source_map else None
                    errors.append(
                        SemanticError(
                            code="UNKNOWN_MEASURE",
                            message=(
                                f"Window metric '{name}' references "
                                f"unknown measure '{ref_measure}'"
                            ),
                            path=f"metrics.{name}.measure",
                            span=span,
                        )
                    )

                win_time_dim = raw_metric.get("timeDimension", "")
                if win_time_dim and win_time_dim not in dimensions:
                    span = (
                        source_map.get(f"metrics.{name}.timeDimension") if source_map else None
                    )
                    errors.append(
                        SemanticError(
                            code="WINDOW_UNKNOWN_TIME_DIMENSION",
                            message=(
                                f"Window metric '{name}' references "
                                f"unknown time dimension '{win_time_dim}'"
                            ),
                            path=f"metrics.{name}.timeDimension",
                            span=span,
                            suggestions=_suggest_similar(win_time_dim, list(dimensions.keys())),
                        )
                    )

                metrics[name] = Metric(
                    name=name,
                    type=MetricType.WINDOW,
                    measure=ref_measure,
                    time_dimension=raw_metric.get("timeDimension"),
                    window_function=raw_metric.get("windowFunction"),
                    offset=raw_metric.get("offset"),
                    buckets=raw_metric.get("buckets"),
                    order_direction=raw_metric.get("orderDirection", "desc"),
                    default_value=raw_metric.get("defaultValue"),
                    partition_by=list(raw_metric.get("partitionBy", []) or []),
                    data_type=raw_metric.get("dataType"),
                    description=raw_metric.get("description"),
                    format=raw_metric.get("format"),
                    owner=raw_metric.get("owner"),
                    synonyms=raw_metric.get("synonyms", []),
                    custom_extensions=_parse_extensions(raw_metric),
                    external_concept_mappings=concept_links,
                )
            else:
                # Derived metric (default). It may reference another derived
                # metric, expanded in place down to real aggregates, or a
                # window metric, which the window wrapper projects as a
                # column of its base CTE — that is what makes
                # ``{[Revenue]} - {[Revenue Prior Month]}`` work, or a
                # reaggregate metric, which reaggregate_wrap rebuilds the
                # formula over.
                expression = raw_metric.get("expression", "")
                self._validate_metric_expression_refs(
                    name,
                    expression,
                    measures,
                    errors,
                    source_map,
                    metrics,
                    synthesized_measure_names,
                    composable_metric_types=(
                        MetricType.DERIVED,
                        MetricType.WINDOW,
                        MetricType.REAGGREGATE,
                    ),
                )

                metrics[name] = Metric(
                    name=name,
                    expression=expression,
                    data_type=raw_metric.get("dataType"),
                    description=raw_metric.get("description"),
                    format=raw_metric.get("format"),
                    owner=raw_metric.get("owner"),
                    synonyms=raw_metric.get("synonyms", []),
                    custom_extensions=_parse_extensions(raw_metric),
                    external_concept_mappings=concept_links,
                )
        except Exception as e:
            span = source_map.get(f"metrics.{name}") if source_map else None
            errors.append(
                SemanticError(
                    code="METRIC_PARSE_ERROR",
                    message=f"Failed to parse metric '{name}': {e}",
                    path=f"metrics.{name}",
                    span=span,
                )
            )

    # Parse static model filters
    model_filters: list[ModelFilter] = []
    raw_filters = raw.get("filters", [])
    if not isinstance(raw_filters, list):
        errors.append(
            SemanticError(
                code="FILTER_PARSE_ERROR",
                message="'filters' must be a YAML list, not a mapping or scalar",
                path="filters",
            )
        )
        raw_filters = []
    for i, rf in enumerate(raw_filters):
        try:
            _check_unknown_keys(rf, _MODEL_FILTER_KEYS, f"filters[{i}]", errors, source_map)
            obj_name = rf.get("dataObject", "")
            col_name = rf.get("column", "")
            if obj_name and obj_name not in data_objects:
                span = source_map.get(f"filters[{i}]") if source_map else None
                errors.append(
                    SemanticError(
                        code="UNKNOWN_FILTER_DATA_OBJECT",
                        message=(
                            f"Static filter[{i}] references unknown data object '{obj_name}'"
                        ),
                        path=f"filters[{i}]",
                        span=span,
                    )
                )
            elif obj_name and col_name and col_name not in data_objects[obj_name].columns:
                span = source_map.get(f"filters[{i}]") if source_map else None
                errors.append(
                    SemanticError(
                        code="UNKNOWN_FILTER_COLUMN",
                        message=(
                            f"Static filter[{i}] references unknown column "
                            f"'{col_name}' in data object '{obj_name}'"
                        ),
                        path=f"filters[{i}]",
                        span=span,
                    )
                )
            raw_val = rf.get("value")
            raw_vals = rf.get("values", [])
            model_filters.append(
                ModelFilter(
                    data_object=obj_name,
                    column=col_name,
                    operator=rf.get("operator", "equals"),
                    value=_coerce_filter_value(raw_val),
                    values=[_coerce_filter_value(v) for v in raw_vals],
                )
            )
        except Exception as e:
            span = source_map.get(f"filters[{i}]") if source_map else None
            errors.append(
                SemanticError(
                    code="FILTER_PARSE_ERROR",
                    message=f"Failed to parse static filter[{i}]: {e}",
                    path=f"filters[{i}]",
                    span=span,
                )
            )

    settings = _parse_settings(raw.get("settings"), errors, source_map)

    rules = self._parse_rules(
        raw.get("rules"),
        dimensions,
        set(measures) | synthesized_measure_names,
        set(metrics),
        prefixes,
        errors,
        source_map,
    )

    # Parse examples block (PLAN_agent_api_improvements §5)
    examples = self._parse_examples(raw.get("examples"), errors)

    model = SemanticModel(
        version=raw.get("version", 1.0),
        name=raw.get("name"),
        description=raw.get("description"),
        data_objects=data_objects,
        dimensions=dimensions,
        measures=measures,
        metrics=metrics,
        filters=model_filters,
        rules=rules,
        examples=examples,
        extends_sources=raw.get("_extends_sources", []),
        inherits_source=raw.get("_inherits_source"),
        owner=raw.get("owner"),
        # Sanitized above so an invalid value is a structured error, not a
        # ValidationError raised here.
        expose_counts=_expose_counts,
        count_label_pattern=_count_pattern,
        custom_extensions=_parse_extensions(raw, "", errors, source_map),
        ontology=ontology,
        external_concept_mappings=_parse_concept_mappings(
            raw, "", prefixes, errors, source_map
        ),
        settings=settings,
    )

    result = ValidationResult(
        valid=len(errors) == 0,
        errors=errors,
        warnings=warnings,
    )

    return model, result