fix(dbt): substitute DERIVED metric references in a single pass

converters/dbt/README.md

@@ -124,6 +124,7 @@ manifest_json = result.output.model_dump_json(by_alias=True, exclude_none=True,
 | `PRIVATE_METRIC_DROPPED` | Ossie has no visibility modifiers |
 | `NATURAL_ENTITY_DROPPED` | Ossie has no natural-key entity type |
 | `CUMULATIVE_SEMANTICS_LOSS` | Window/grain semantics cannot be expressed in an Ossie expression string; the base aggregation is preserved |
+| `AMBIGUOUS_REFERENCE_METRIC_DROPPED` | An input metric is listed more than once under one reference with differing filters, so the expression reference is ambiguous; give each occurrence a distinct alias |
 
 **Ossie → MSI** reconstructs a best-effort MSI manifest from Ossie's simpler schema. Nothing is dropped for supported inputs, but Ossie carries less structural information than MSI, so the converter makes the following choices:
 

converters/dbt/src/ossie_dbt/cli.py

@@ -40,12 +40,17 @@
     ConverterIssueType.PRIVATE_METRIC_DROPPED: "Ossie has no visibility modifiers",
     ConverterIssueType.NATURAL_ENTITY_DROPPED: "Ossie has no natural-key entity type",
     ConverterIssueType.CUMULATIVE_SEMANTICS_LOSS: "Ossie expressions cannot represent window or grain semantics; the base aggregation was preserved",
+    ConverterIssueType.AMBIGUOUS_REFERENCE_METRIC_DROPPED: (
+        "an input metric is listed more than once under one reference with differing filters, "
+        "so the expression reference is ambiguous; give each occurrence a distinct alias"
+    ),
 }
 
 _DROPPED_ISSUE_TYPES = {
     ConverterIssueType.CONVERSION_METRIC_DROPPED,
     ConverterIssueType.PRIVATE_METRIC_DROPPED,
     ConverterIssueType.NATURAL_ENTITY_DROPPED,
+    ConverterIssueType.AMBIGUOUS_REFERENCE_METRIC_DROPPED,
 }
 
 

converters/dbt/src/ossie_dbt/converter_issues.py

@@ -27,6 +27,7 @@ class ConverterIssueType(Enum):
     PRIVATE_METRIC_DROPPED = "PRIVATE_METRIC_DROPPED"
     NATURAL_ENTITY_DROPPED = "NATURAL_ENTITY_DROPPED"
     CUMULATIVE_SEMANTICS_LOSS = "CUMULATIVE_SEMANTICS_LOSS"
+    AMBIGUOUS_REFERENCE_METRIC_DROPPED = "AMBIGUOUS_REFERENCE_METRIC_DROPPED"
 
 
 @dataclass(frozen=True)

converters/dbt/src/ossie_dbt/filter_utils.py

@@ -154,8 +154,14 @@ def _collect_filter_sql(*filters: Optional[WhereFilterIntersection]) -> Optional
 
 
 def _merge_filter_sqls(*parts: Optional[str]) -> Optional[str]:
-    """Join non-None SQL filter strings with AND, wrapping each in parens when multiple."""
-    active = [p for p in parts if p]
+    """Join non-None SQL filter strings with AND, wrapping each in parens when multiple.
+
+    Repeated fragments are dropped. Filters are conjoined and AND is idempotent, so an
+    input metric that restates a filter an enclosing metric already applies contributes
+    nothing but ``(X) AND (X)`` noise — and would otherwise make two logically identical
+    resolutions of the same reference compare unequal.
+    """
+    active = list(dict.fromkeys(p for p in parts if p))
     if not active:
         return None
     if len(active) == 1:

converters/dbt/src/ossie_dbt/msi_to_ossie.py

@@ -72,6 +72,14 @@ class _RelationshipDirection:
     to_col: str
 
 
+class AmbiguousDerivedReferenceError(Exception):
+    """A DERIVED metric's expression reference does not resolve to a single expression.
+
+    Raised by the expression resolver and handled in :meth:`MSIToOssieConverter.convert`,
+    which drops the metric and records a ConverterIssue rather than failing the run.
+    """
+
+
 class MSIToOssieConverter:
     """Converts an MSI SemanticManifest into an Ossie Document."""
 
@@ -109,7 +117,18 @@ def convert(
                 issues.append(
                     ConverterIssue(issue_type=ConverterIssueType.CUMULATIVE_SEMANTICS_LOSS, element_name=metric.name)
                 )
-            expr = self._resolve_metric_expression(metric, metric_index, expression_cache)
+            try:
+                expr = self._resolve_metric_expression(metric, metric_index, expression_cache)
+            except AmbiguousDerivedReferenceError:
+                # Every other unsupported shape drops one metric and records an issue;
+                # an ambiguous reference is no reason to fail the whole conversion.
+                issues.append(
+                    ConverterIssue(
+                        issue_type=ConverterIssueType.AMBIGUOUS_REFERENCE_METRIC_DROPPED,
+                        element_name=metric.name,
+                    )
+                )
+                continue
             ossie_metrics.append(
                 OssieMetric(
                     name=metric.name,
@@ -332,17 +351,58 @@ def _resolve_derived(
         """Resolve a DERIVED metric by substituting each input metric's expression into the expr string.
 
         Compound sub-expressions (DERIVED/RATIO) are wrapped in parentheses to preserve operator precedence.
+
+        All references are substituted in a single pass. Substituting them one at a
+        time would re-scan text inserted by an earlier reference, so a metric named
+        after a column appearing in an already-inlined expression would be expanded
+        twice. The replacement is a callback rather than a string so that backslashes
+        in the resolved SQL (e.g. from a `LIKE 'a\\b'` filter) are inserted verbatim
+        instead of being interpreted as `re.sub` template escapes.
+
+        Listing the same input metric twice under one reference raises
+        :class:`AmbiguousDerivedReferenceError` when the expression uses that reference
+        and the two occurrences resolve differently (e.g. distinct per-input filters and
+        no aliases): the expression has a single token for them, so either resolution
+        would be an arbitrary choice. MetricFlow does not reject this shape upstream —
+        `DerivedMetricRule._validate_alias_collision` only compares entries that set an
+        alias. Occurrences that resolve identically are redundant rather than ambiguous,
+        and a duplicated reference the expression never substitutes cannot affect the
+        result; both are accepted.
         """
         expr = metric.type_params.expr or ""
+        resolutions: Dict[str, List[str]] = {}
         for input_metric in metric.type_params.metrics or []:
             ref = input_metric.alias if input_metric.alias else input_metric.name
             dep_metric = self._lookup_metric(metric_index, input_metric.name, f"DERIVED metric '{metric.name}'")
             input_filter = _merge_filter_sqls(filter_sql, _collect_filter_sql(input_metric.filter))
             resolved = self._resolve_metric_expression(dep_metric, metric_index, cache, input_filter)
             if dep_metric.type in (MetricType.DERIVED, MetricType.RATIO):
                 resolved = f"({resolved})"
-            expr = re.sub(rf"\b{re.escape(ref)}\b", resolved, expr)
-        return expr
+            distinct = resolutions.setdefault(ref, [])
+            if resolved not in distinct:
+                distinct.append(resolved)
+
+        if not resolutions:
+            return expr
+
+        # The `\b` anchors already stop a short reference from matching inside a
+        # longer identifier; sorting by length (then name) keeps the alternation order stable
+        # and independent of the order metrics happen to be declared in.
+        pattern = re.compile(
+            r"\b(" + "|".join(re.escape(ref) for ref in sorted(resolutions, key=lambda ref: (-len(ref), ref))) + r")\b"
+        )
+
+        # Only a reference the expression actually uses can be ambiguous; one that is
+        # listed but never substituted has no bearing on the result.
+        for ref in sorted(set(pattern.findall(expr))):
+            if len(resolutions[ref]) > 1:
+                raise AmbiguousDerivedReferenceError(
+                    "DERIVED metric references an input metric that is listed more than once with "
+                    "differing resolutions, making the reference ambiguous; give each occurrence a "
+                    f"distinct alias: metric_name={metric.name!r}, reference={ref!r}"
+                )
+
+        return pattern.sub(lambda match: resolutions[match.group(0)][0], expr)
 
     @staticmethod
     def _build_entity_index(

converters/dbt/tests/test_msi_to_ossie.py

@@ -686,6 +686,214 @@ def test_derived_metric_uses_alias_for_substitution(self) -> None:
         profit_ossie = next(m for m in _ossie_metrics(result) if m.name == "profit")
         assert profit_ossie.expression.dialects[0].expression == "SUM(orders.amount) - SUM(orders.cost_amount)"
 
+    def test_derived_metric_does_not_re_expand_an_inlined_reference(self) -> None:
+        """A reference is substituted once, even when its name appears in already-inlined text.
+
+        `gross` inlines to `SUM(orders.net)`, which contains the name of the second
+        input metric. Substituting references one at a time would expand `net` inside
+        that text as well.
+        """
+        sm = semantic_model_with_guaranteed_meta(
+            name="orders",
+            measures=[
+                _measure("gross", agg=AggregationType.SUM, expr="net"),
+                _measure("net", agg=AggregationType.SUM, expr="net_amount"),
+            ],
+        )
+        gross_m = _simple_metric("gross", "gross")
+        net_m = _simple_metric("net", "net")
+        margin = PydanticMetric(
+            name="margin",
+            description=None,
+            type=MetricType.DERIVED,
+            type_params=PydanticMetricTypeParams(
+                expr="gross - net",
+                metrics=[
+                    PydanticMetricInput(name="gross"),
+                    PydanticMetricInput(name="net"),
+                ],
+            ),
+            filter=None,
+            metadata=default_meta(),
+            config=None,
+        )
+        result = (
+            MSIToOssieConverter().convert(_manifest(semantic_models=[sm], metrics=[gross_m, net_m, margin])).output
+        )
+
+        margin_ossie = next(m for m in _ossie_metrics(result) if m.name == "margin")
+        assert margin_ossie.expression.dialects[0].expression == "SUM(orders.net) - SUM(orders.net_amount)"
+
+    def test_derived_metric_preserves_backslashes_from_a_filter(self) -> None:
+        """Backslashes in an inlined expression survive substitution verbatim.
+
+        A resolved expression is inserted as a literal, not as a `re.sub` replacement
+        template, so an escape sequence such as `\\b` in a filter's SQL is not
+        reinterpreted (`\\b` would otherwise become a backspace character).
+        """
+        sm = semantic_model_with_guaranteed_meta(
+            name="orders",
+            measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")],
+        )
+        revenue_m = _simple_metric("revenue", "revenue")
+        revenue_m.filter = _filter(r"{{ Dimension('order__path') }} LIKE 'a\b'")
+        scaled = PydanticMetric(
+            name="scaled",
+            description=None,
+            type=MetricType.DERIVED,
+            type_params=PydanticMetricTypeParams(
+                expr="revenue * 2",
+                metrics=[PydanticMetricInput(name="revenue")],
+            ),
+            filter=None,
+            metadata=default_meta(),
+            config=None,
+        )
+        result = MSIToOssieConverter().convert(_manifest(semantic_models=[sm], metrics=[revenue_m, scaled])).output
+
+        revenue_ossie = next(m for m in _ossie_metrics(result) if m.name == "revenue")
+        scaled_ossie = next(m for m in _ossie_metrics(result) if m.name == "scaled")
+        assert revenue_ossie.expression.dialects[0].expression == (
+            r"SUM(CASE WHEN order__path LIKE 'a\b' THEN orders.amount END)"
+        )
+        assert scaled_ossie.expression.dialects[0].expression == (
+            r"SUM(CASE WHEN order__path LIKE 'a\b' THEN orders.amount END) * 2"
+        )
+
+    def test_derived_metric_drops_a_used_reference_listed_twice_with_differing_filters(self) -> None:
+        """An input metric listed twice under one used reference, resolving differently, is ambiguous.
+
+        MetricFlow accepts this shape — `DerivedMetricRule._validate_alias_collision`
+        only compares entries that set an alias — so the converter has to handle it. It
+        drops the metric and records an issue, as it does for every other unsupported
+        shape, rather than failing the whole conversion.
+        """
+        sm = semantic_model_with_guaranteed_meta(
+            name="orders",
+            measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")],
+        )
+        revenue_m = _simple_metric("revenue", "revenue")
+        both = PydanticMetric(
+            name="both",
+            description=None,
+            type=MetricType.DERIVED,
+            type_params=PydanticMetricTypeParams(
+                expr="revenue",
+                metrics=[
+                    PydanticMetricInput(name="revenue", filter=_filter("{{ Dimension('order__region') }} = 'EU'")),
+                    PydanticMetricInput(name="revenue", filter=_filter("{{ Dimension('order__region') }} = 'US'")),
+                ],
+            ),
+            filter=None,
+            metadata=default_meta(),
+            config=None,
+        )
+        result = MSIToOssieConverter().convert(_manifest(semantic_models=[sm], metrics=[revenue_m, both]))
+
+        assert [m.name for m in _ossie_metrics(result.output)] == ["revenue"]
+        assert len(result.issues) == 1
+        assert result.issues[0].issue_type == ConverterIssueType.AMBIGUOUS_REFERENCE_METRIC_DROPPED
+        assert result.issues[0].element_name == "both"
+
+    def test_derived_metric_accepts_a_duplicate_reference_the_expression_never_uses(self) -> None:
+        """A reference that is never substituted cannot be ambiguous, however it resolves."""
+        sm = semantic_model_with_guaranteed_meta(
+            name="orders",
+            measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")],
+        )
+        revenue_m = _simple_metric("revenue", "revenue")
+        constant = PydanticMetric(
+            name="constant",
+            description=None,
+            type=MetricType.DERIVED,
+            type_params=PydanticMetricTypeParams(
+                expr="1 + 1",
+                metrics=[
+                    PydanticMetricInput(name="revenue", filter=_filter("{{ Dimension('order__region') }} = 'EU'")),
+                    PydanticMetricInput(name="revenue", filter=_filter("{{ Dimension('order__region') }} = 'US'")),
+                ],
+            ),
+            filter=None,
+            metadata=default_meta(),
+            config=None,
+        )
+        result = MSIToOssieConverter().convert(_manifest(semantic_models=[sm], metrics=[revenue_m, constant]))
+
+        assert next(m for m in _ossie_metrics(result.output) if m.name == "constant").expression.dialects[
+            0
+        ].expression == "1 + 1"
+        assert result.issues == []
+
+    def test_derived_metric_accepts_a_duplicate_that_restates_the_enclosing_filter(self) -> None:
+        """Restating a filter the enclosing metric already applies is redundant, not ambiguous.
+
+        AND is idempotent, so the bare occurrence and the one repeating the parent's own
+        filter describe the same set of rows and must not be treated as two resolutions.
+        """
+        region_eu = "{{ Dimension('order__region') }} = 'EU'"
+        sm = semantic_model_with_guaranteed_meta(
+            name="orders",
+            measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")],
+        )
+        revenue_m = _simple_metric("revenue", "revenue")
+        scaled = PydanticMetric(
+            name="scaled",
+            description=None,
+            type=MetricType.DERIVED,
+            type_params=PydanticMetricTypeParams(
+                expr="revenue * 2",
+                metrics=[
+                    PydanticMetricInput(name="revenue"),
+                    PydanticMetricInput(name="revenue", filter=_filter(region_eu)),
+                ],
+            ),
+            filter=_filter(region_eu),
+            metadata=default_meta(),
+            config=None,
+        )
+        result = MSIToOssieConverter().convert(_manifest(semantic_models=[sm], metrics=[revenue_m, scaled]))
+
+        scaled_ossie = next(m for m in _ossie_metrics(result.output) if m.name == "scaled")
+        assert scaled_ossie.expression.dialects[0].expression == (
+            "SUM(CASE WHEN order__region = 'EU' THEN orders.amount END) * 2"
+        )
+        assert result.issues == []
+
+    def test_derived_metric_accepts_a_reference_listed_twice_resolving_identically(self) -> None:
+        """A redundant duplicate is not ambiguous: both occurrences resolve to the same SQL."""
+        sm = semantic_model_with_guaranteed_meta(
+            name="orders",
+            measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")],
+        )
+        revenue_m = _simple_metric("revenue", "revenue")
+        doubled = PydanticMetric(
+            name="doubled",
+            description=None,
+            type=MetricType.DERIVED,
+            type_params=PydanticMetricTypeParams(
+                expr="revenue + revenue",
+                metrics=[
+                    PydanticMetricInput(name="revenue"),
+                    PydanticMetricInput(name="revenue"),
+                ],
+            ),
+            filter=None,
+            metadata=default_meta(),
+            config=None,
+        )
+        result = (
+            MSIToOssieConverter().convert(_manifest(semantic_models=[sm], metrics=[revenue_m, doubled])).output
+        )
+
+        doubled_ossie = next(m for m in _ossie_metrics(result) if m.name == "doubled")
+        assert doubled_ossie.expression.dialects[0].expression == "SUM(orders.amount) + SUM(orders.amount)"
+
+    def test_every_issue_type_has_a_cli_reason(self) -> None:
+        """The CLI looks each issue type up by key, so an unmapped one would crash the run."""
+        from ossie_dbt.cli import _ISSUE_REASON
+
+        assert set(_ISSUE_REASON) == set(ConverterIssueType)
+
     def test_derived_metric_nested(self, snapshot: SnapshotAssertion) -> None:
         sm = semantic_model_with_guaranteed_meta(
             name="orders",