fix(relations): gate against authoritative input dialect, not anchor (item 95)

src/mountainash/relations/core/unified_visitor/relation_visitor.py

@@ -88,6 +88,24 @@ def _present_expression_function_keys(node: RelationNode, op: Any) -> frozenset:
     return frozenset(keys)
 
 
+_UNRESOLVED = object()
+
+
+def _first_input_node(node: RelationNode) -> "RelationNode | None":
+    from mountainash.relations.core.relation_nodes import (
+        AggregateRelNode, ConformRelNode, ExtensionRelNode, FetchRelNode,
+        FilterRelNode, JoinRelNode, ProjectRelNode, SetRelNode, SortRelNode,
+    )
+    if isinstance(node, JoinRelNode):
+        return node.left
+    if isinstance(node, SetRelNode):
+        return node.inputs[0] if node.inputs else None
+    if isinstance(node, (FilterRelNode, ProjectRelNode, SortRelNode, FetchRelNode,
+                         AggregateRelNode, ConformRelNode, ExtensionRelNode)):
+        return node.input
+    return None
+
+
 class UnifiedRelationVisitor:
     """Walks a relational AST and produces backend-native results.
 
@@ -103,6 +121,7 @@ def __init__(
         *,
         ref_resolver: Optional[Callable[[str], Any]] = None,
         key_context: Optional["KeyDriftContext"] = None,
+        identity_resolver: Optional[Callable[[str], Any]] = None,
         enforce_capabilities: bool = True,
     ) -> None:
         self.backend = relation_system
@@ -114,6 +133,7 @@ def __init__(
         # then never assesses the keys dimension and ConformDrift.key_changes
         # stays None (not assessed).
         self.key_context = key_context
+        self.identity_resolver = identity_resolver
         self.enforce_capabilities = enforce_capabilities
         if enforce_capabilities:
             # A gating consumer must ensure the capability declaration modules
@@ -174,7 +194,9 @@ def _gate_capabilities(self, node, op) -> None:
         family = getattr(self.backend, "backend_type", None)
         if family is None:
             return
-        dialect = getattr(self.backend, "dialect", None)
+        dialect = self._authoritative_dialect(node, op)
+        if dialect is _UNRESOLVED:
+            dialect = getattr(self.backend, "dialect", None)
 
         def _raise(fact):
             raise BackendCapabilityError(
@@ -208,6 +230,40 @@ def _raise(fact):
             if getattr(node, param, None) is not None:
                 _raise(fact)
 
+    def _authoritative_dialect(self, node: RelationNode, op: Any):
+        input_node = _first_input_node(node)
+        if input_node is None:
+            return _UNRESOLVED
+        family, dialect = self._physical_identity(input_node)
+        if family is None or family != self.backend.backend_type:
+            return _UNRESOLVED
+        return dialect
+
+    def _physical_identity(self, node: RelationNode, seen: "set | None" = None):
+        if seen is None:
+            seen = set()
+        try:
+            from mountainash.core.backend_detection import identify_backend_identity
+            from mountainash.relations.core.relation_nodes import ReadRelNode
+            from mountainash.relations.core.relation_nodes.extensions_mountainash import (
+                RefRelNode, ResourceReadRelNode, SourceRelNode,
+            )
+            if isinstance(node, ReadRelNode):
+                ident = identify_backend_identity(node.dataframe)
+                return ident.family, ident.dialect
+            if isinstance(node, RefRelNode):
+                if self.identity_resolver is None or node.name in seen:
+                    return None, None
+                resolved = self.identity_resolver(node.name)
+                return (self._physical_identity(resolved, seen | {node.name})
+                        if resolved is not None else (None, None))
+            if isinstance(node, (SourceRelNode, ResourceReadRelNode)):
+                return None, None
+            child = _first_input_node(node)
+            return self._physical_identity(child, seen) if child is not None else (None, None)
+        except Exception:
+            return None, None
+
     def _dispatch(self, node: RelationNode, op: Any) -> Any:
         if self.enforce_capabilities:
             self._gate_capabilities(node, op)

src/mountainash/relations/dag/dag.py

@@ -412,6 +412,7 @@ def resolver(n: str) -> Any:
             expression_visitor=expr_visitor,
             ref_resolver=resolver,
             key_context=key_context,
+            identity_resolver=lambda name: self.relations[name]._node,
         )
 
         # Compile refs in topological order

tests/relations/test_rel_authoritative_dialect_gating.py

@@ -0,0 +1,164 @@
+"""Input-authoritative dialect gating for the capability gate (item 95).
+
+_gate_capabilities keys its dialect-scoped lookup off the visitor's anchor
+dialect, so a GATE fact scoped to an operation's authoritative (left) input
+dialect fires against the wrong string. This item resolves the authoritative
+input dialect recursively.
+
+Design: mountainash-central
+2026-08-14-multi-input-node-anchor-dialect-gating-design.md (Revision 5).
+"""
+from __future__ import annotations
+
+import narwhals as nw
+import pandas as pd
+import polars as pl
+import pytest
+
+import mountainash as ma
+from mountainash.core.capabilities import (
+    CapabilityFact,
+    CapabilityLevel,
+    CapabilityRegistry,
+    Enforcement,
+    WILDCARD_PARAM,
+)
+from mountainash.core.constants import CONST_BACKEND, SetType
+from mountainash.core.types import BackendCapabilityError
+from mountainash.relations.core.relation_system.relation_keys.enums import (
+    RKEY_SUBSTRAIT_REL,
+)
+from mountainash.relations.dag import RelationDAG
+
+import mountainash.relations.backends  # noqa: F401
+import mountainash.expressions.backends  # noqa: F401
+
+
+def _nw_polars(data: dict):
+    return nw.from_native(pl.DataFrame(data), eager_only=True)
+
+
+def _nw_pandas(data: dict):
+    return nw.from_native(pd.DataFrame(data), eager_only=True)
+
+
+@pytest.fixture
+def _narwhals_pandas_filter_gate_fact():
+    snap = CapabilityRegistry.snapshot()
+    try:
+        CapabilityRegistry.register_backend(
+            CONST_BACKEND.NARWHALS,
+            [
+                CapabilityFact(
+                    operation_key=RKEY_SUBSTRAIT_REL.FILTER,
+                    param=WILDCARD_PARAM,
+                    level=CapabilityLevel.UNSUPPORTED,
+                    backend=CONST_BACKEND.NARWHALS,
+                    dialect="narwhals-pandas",
+                    since="2026-08-14",
+                    message="test-only BUILD-time gate for narwhals-pandas filter",
+                    enforcement=Enforcement.GATE,
+                )
+            ],
+        )
+        yield
+    finally:
+        CapabilityRegistry.restore(snap)
+
+
+class TestInlineOperandDialectGate:
+    def test_filter_gate_fires_on_inline_left_operand_dialect(
+        self, _narwhals_pandas_filter_gate_fact
+    ):
+        # Anchor is narwhals-polars (a_polars is alphabetically first); the
+        # Filter is INLINE in the target tree (not a separately-compiled dep),
+        # so its gate runs under the anchor pair. Its input is a narwhals-pandas
+        # ref, so the gate must fire against narwhals-pandas -- today it uses
+        # the anchor (narwhals-polars) and does NOT fire.
+        dag = RelationDAG()
+        dag.add("a_polars", ma.relation(_nw_polars({"k": [1, 2]})))
+        dag.add("z_pandas", ma.relation(_nw_pandas({"k": [1, 2]})))
+        dag.add(
+            "target",
+            dag.ref("z_pandas").filter(ma.col("k") > 0).join(dag.ref("a_polars"), on="k"),
+        )
+        with pytest.raises(BackendCapabilityError):
+            dag.collect("target")
+
+
+class TestAuthoritativeDialectCases:
+    def test_gate_does_not_fire_when_left_matches_anchor(self, _narwhals_pandas_filter_gate_fact):
+        dag = RelationDAG()
+        dag.add("a_polars", ma.relation(_nw_polars({"k": [1, 2]})))
+        dag.add("z_polars2", ma.relation(_nw_polars({"k": [1, 2]})))
+        dag.add("target", dag.ref("z_polars2").filter(ma.col("k") > 0).join(dag.ref("a_polars"), on="k"))
+        dag.collect("target")   # narwhals-pandas fact must NOT fire
+
+    def test_join_gate_fires_on_left_operand_dialect(self):
+        snap = CapabilityRegistry.snapshot()
+        try:
+            CapabilityRegistry.register_backend(CONST_BACKEND.NARWHALS, [
+                CapabilityFact(
+                    operation_key=RKEY_SUBSTRAIT_REL.JOIN, param=WILDCARD_PARAM,
+                    level=CapabilityLevel.UNSUPPORTED, backend=CONST_BACKEND.NARWHALS,
+                    dialect="narwhals-pandas", since="2026-08-14",
+                    enforcement=Enforcement.GATE,
+                    message="join gate on narwhals-pandas",
+                )
+            ])
+            dag = RelationDAG()
+            dag.add("a_polars", ma.relation(_nw_polars({"k": [1, 2]})))
+            dag.add("z_pandas", ma.relation(_nw_pandas({"k": [1, 2]})))
+            dag.add("target", dag.ref("z_pandas").join(dag.ref("a_polars"), on="k"))
+            with pytest.raises(BackendCapabilityError):
+                dag.collect("target")
+        finally:
+            CapabilityRegistry.restore(snap)
+
+    def test_unbound_ibis_input_yields_none(self):
+        import ibis
+        from mountainash.relations.core.unified_visitor.relation_visitor import (
+            UnifiedRelationVisitor,
+        )
+        from mountainash.relations.core.relation_nodes import ReadRelNode
+
+        class _FakeBackend:
+            backend_type = CONST_BACKEND.IBIS
+            dialect = "ibis-duckdb"   # anchor has a KNOWN dialect
+
+        ib = ibis.memtable({"k": [1]})   # unbound -> (IBIS, None)
+        visitor = UnifiedRelationVisitor(
+            _FakeBackend(), expression_visitor=None, identity_resolver=None
+        )
+        family, dialect = visitor._physical_identity(ReadRelNode(dataframe=ib))
+        assert family is CONST_BACKEND.IBIS
+        assert dialect is None   # explicitly unknown, NOT the anchor's "ibis-duckdb"
+
+    def test_cycle_protection_returns_unresolved(self):
+        from mountainash.relations.core.unified_visitor.relation_visitor import (
+            UnifiedRelationVisitor,
+        )
+        from mountainash.relations.core.relation_nodes.extensions_mountainash import (
+            RefRelNode,
+        )
+
+        class _FakeBackend:
+            backend_type = CONST_BACKEND.NARWHALS
+            dialect = "narwhals-pandas"
+
+        nodes = {"a": RefRelNode(name="b"), "b": RefRelNode(name="a")}
+        visitor = UnifiedRelationVisitor(
+            _FakeBackend(), expression_visitor=None,
+            identity_resolver=lambda name: nodes[name],
+        )
+        family, dialect = visitor._physical_identity(nodes["a"])
+        assert family is None and dialect is None   # cycle -> unresolved, no recursion
+
+    def test_empty_set_node_no_indexerror(self):
+        from mountainash.relations.core.unified_visitor.relation_visitor import (
+            _first_input_node,
+        )
+        from mountainash.relations.core.relation_nodes import SetRelNode
+
+        node = SetRelNode(inputs=[], set_type=SetType.UNION_ALL)
+        assert _first_input_node(node) is None   # no IndexError on inputs[0]

tests/relations/test_rel_multi_input_dialect_coercion.py

@@ -416,36 +416,28 @@ def _narwhals_pandas_join_gate_fact():
         CapabilityRegistry.restore(snap)
 
 
-class TestGatingUsesAnchorDialectNotTrueLeftOperandDialect:
-    """Design spec testing plan #10 (required per Codex's suggested
-    resolution for round-1 finding #2 -- this item ships the fix, item 95
-    ships the gating-precision fix, sequenced after). Pins the current,
-    known-limited behaviour with a REAL dialect-scoped CapabilityFact,
-    not merely an inspection of visitor.backend.dialect."""
-
-    def test_pandas_scoped_join_gate_does_not_fire_when_anchor_is_polars(
+class TestGatingUsesAuthoritativeDialectNotAnchor:
+    """Item 95: _gate_capabilities resolves the operation's authoritative
+    (first/left) input dialect, not the visitor's anchor dialect. This test
+    was item 91 testing plan #10, which pinned the OLD anchor-dialect
+    limitation; item 95 ships the gating-precision fix and inverts it."""
+
+    def test_pandas_scoped_join_gate_fires_when_left_operand_is_pandas(
         self, _narwhals_pandas_join_gate_fact
     ):
         from mountainash.relations.dag import RelationDAG
+        from mountainash.core.types import BackendCapabilityError
 
         dag = RelationDAG()
         # "a_polars_src" sorts alphabetically first -> becomes the anchor
-        # (RelationDAG._execute_with_visitor's own documented selection:
-        # sorted(all_refs)[0]) -- regardless of tree position.
+        # (narwhals-polars), regardless of tree position. The join's TRUE
+        # left/authoritative operand is "z_pandas_src" (narwhals-pandas) --
+        # exactly what the registered GATE fact targets.
         dag.add("a_polars_src", ma.relation(_nw_polars({"id": [1], "x": [1]})))
         dag.add("z_pandas_src", ma.relation(_nw_pandas({"id": [1], "y": [1]})))
-        # The join's TRUE left/authoritative operand is "z_pandas_src"
-        # (narwhals-pandas) -- exactly what the registered GATE fact
-        # targets.
         joined = dag.ref("z_pandas_src").join(dag.ref("a_polars_src"), on="id")
 
-        result, visitor = dag._execute_with_visitor(joined)
-
-        # Pins the current, documented limitation (item 95's charter):
-        # the anchor is narwhals-polars, NOT the join's true left operand
-        # (narwhals-pandas) -- so the pandas-scoped GATE fact never fires,
-        # even though a fully operand-aware gate SHOULD have blocked this
-        # join. The join succeeds anyway (coercion, Task 3, still works
-        # correctly regardless of gating precision).
-        assert visitor.backend.dialect == "narwhals-polars"
-        assert result is not None
+        # Item 95: the gate now fires against the authoritative left operand's
+        # dialect (narwhals-pandas), not the anchor's (narwhals-polars).
+        with pytest.raises(BackendCapabilityError):
+            dag._execute_with_visitor(joined)