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)

        # 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),
                    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,
                    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),
                )
            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"),
                    within_group=raw_meas.get("withinGroup"),
                    owner=raw_meas.get("owner"),
                    synonyms=raw_meas.get("synonyms", []),
                    custom_extensions=_parse_extensions(raw_meas),
                )
            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")

                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
                    ):
                        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),
                    )
                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),
                    )
                elif metric_type == MetricType.WINDOW:
                    # Window metric (rank/lag/lead/ntile/first_value/last_value)
                    ref_measure = raw_metric.get("measure")
                    if (
                        ref_measure
                        and ref_measure not in measures
                        and ref_measure not in synthesized_measure_names
                    ):
                        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),
                    )
                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.
                    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),
                    )

                    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),
                    )
            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)

        # 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,
            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),
            settings=settings,
        )

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

        return model, result

    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,
                )
            )

    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)

    # 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),
                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,
                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),
            )
        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"),
                within_group=raw_meas.get("withinGroup"),
                owner=raw_meas.get("owner"),
                synonyms=raw_meas.get("synonyms", []),
                custom_extensions=_parse_extensions(raw_meas),
            )
        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")

            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
                ):
                    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),
                )
            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),
                )
            elif metric_type == MetricType.WINDOW:
                # Window metric (rank/lag/lead/ntile/first_value/last_value)
                ref_measure = raw_metric.get("measure")
                if (
                    ref_measure
                    and ref_measure not in measures
                    and ref_measure not in synthesized_measure_names
                ):
                    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),
                )
            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.
                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),
                )

                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),
                )
        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)

    # 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,
        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),
        settings=settings,
    )

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

    return model, result