feat(relations): make MATERIALIZE_RESIDUE per-op enrichment live

src/mountainash/core/capabilities/registry.py

@@ -200,6 +200,21 @@ def _validate_fact(family: CONST_BACKEND, fact: CapabilityFact) -> None:
                 "gate_params — the fact could never gate. Add the param to the "
                 "op's gate_params (RelationOperationDef) or declare a non-GATE enforcement role."
             )
+    # Residue reachability (item 98): a handler-routed op's MATERIALIZE_RESIDUE
+    # fact can only ever fire if the handler wraps its native call via
+    # _enrich_native_call (relation_visitor.py) — reject a silently-dead
+    # declaration on an unwrapped handler at registration.
+    if (
+        kind == "relation"
+        and fact.enforcement is Enforcement.MATERIALIZE_RESIDUE
+        and getattr(definition, "handler", None) is not None
+        and not getattr(definition, "wraps_native_call", False)
+    ):
+        raise ValueError(
+            f"CapabilityFact({fact.operation_key!r}): the op is handler-routed "
+            "and does not declare wraps_native_call=True, so a "
+            "MATERIALIZE_RESIDUE fact could never fire through per-op enrichment."
+        )
 
 
 class CapabilityRegistry:

src/mountainash/core/limitations.py

@@ -71,6 +71,7 @@ def enrich_materialization(
     fn: Callable[[], Any],
     *,
     prefer_operation_keys: "frozenset | None" = None,
+    dialect: "str | None" = None,
 ) -> Any:
     """Materialization-boundary enrichment: consult the spine's MATERIALIZE
     residue (matched by native exception type — residue facts keep their
@@ -89,13 +90,18 @@ def enrich_materialization(
             raised error is enriched only when **exactly one** candidate
             matches the exception's type; zero or multiple matches leave
             the original exception to propagate raw rather than guessing.
+        dialect: override the backend's own dialect for the residue lookup
+            — the authoritative input dialect item 95 resolves for
+            multi-input nodes.
     """
     from mountainash.core.capabilities import CapabilityRegistry
     from mountainash.core.types import BackendCapabilityError
 
     family = getattr(backend, "backend_type", None)
     residue = (
-        CapabilityRegistry.residue_for(family, getattr(backend, "dialect", None))
+        CapabilityRegistry.residue_for(
+            family, dialect if dialect is not None else getattr(backend, "dialect", None)
+        )
         if family is not None
         else {}
     )

src/mountainash/relations/core/relation_system/relation_mapping/definitions.py

@@ -135,6 +135,7 @@ def _ext(key, method):
         substrait_rel="JoinRel",
         protocol_method=SubstraitJoinRelationSystemProtocol.join,
         handler=handlers.visit_join,  # cross-backend right-side coercion
+        wraps_native_call=True,
     ),
     RelationOperationDef(
         operation_key=RS.AGGREGATE,
@@ -188,6 +189,7 @@ def _ext(key, method):
         is_extension=True,
         extension_uri=MountainashRelExtension.UTIL,
         handler=handlers.visit_source,
+        wraps_native_call=True,
     ),
     RelationOperationDef(
         operation_key=RM.REF,
@@ -203,13 +205,15 @@ def _ext(key, method):
         extension_uri=MountainashRelExtension.DAG,
         protocol_method=ExtProto.read_resource,
         handler=handlers.visit_resource_read,
+        wraps_native_call=True,
     ),
     RelationOperationDef(
         operation_key=RM.CONFORM,
         node_type=ConformRelNode,
         is_extension=True,
         extension_uri=MountainashRelExtension.CONFORM,
         handler=handlers.visit_conform,
+        wraps_native_call=True,
     ),
     RelationOperationDef(
         operation_key=RM.FETCH_FROM_END,
@@ -229,6 +233,7 @@ def _ext(key, method):
         protocol_method=ExtProto.join_asof,
         handler=handlers.visit_join_asof,
         gate_params=("tolerance",),
+        wraps_native_call=True,
     ),
     RelationOperationDef(
         # No node type: invoked from the conform path, not node dispatch.

src/mountainash/relations/core/relation_system/relation_mapping/handlers.py

@@ -10,29 +10,37 @@
 
 from typing import Any
 
+from mountainash.relations.core.relation_system.relation_keys.enums import (
+    RKEY_MOUNTAINASH_REL,
+    RKEY_SUBSTRAIT_REL,
+)
+
 
 def visit_join(node: Any, visitor: Any) -> Any:
     left = visitor.visit(node.left)
     right = visitor._visit_and_coerce_right(node.right, left)
-    return visitor.backend.join(
-        left, right,
-        join_type=node.join_type,
-        on=node.on,
-        left_on=node.left_on,
-        right_on=node.right_on,
-        suffix=node.suffix,
+    return visitor._enrich_native_call(
+        node, RKEY_SUBSTRAIT_REL.JOIN,
+        lambda: visitor.backend.join(
+            left, right,
+            join_type=node.join_type,
+            on=node.on, left_on=node.left_on,
+            right_on=node.right_on, suffix=node.suffix,
+        ),
     )
 
 
 def visit_join_asof(node: Any, visitor: Any) -> Any:
     left = visitor.visit(node.left)
     right = visitor._visit_and_coerce_right(node.right, left)
-    return visitor.backend.join_asof(
-        left, right,
-        on=node.on[0] if node.on else node.left_on[0],
-        by=node.by,
-        strategy=node.strategy or "backward",
-        tolerance=node.tolerance,
+    return visitor._enrich_native_call(
+        node, RKEY_MOUNTAINASH_REL.JOIN_ASOF,
+        lambda: visitor.backend.join_asof(
+            left, right,
+            on=node.on[0] if node.on else node.left_on[0],
+            by=node.by, strategy=node.strategy or "backward",
+            tolerance=node.tolerance,
+        ),
     )
 
 
@@ -47,21 +55,29 @@ def visit_ref(node: Any, visitor: Any) -> Any:
 
 
 def visit_resource_read(node: Any, visitor: Any) -> Any:
-    out = visitor.backend.read_resource(node.resource)
-    if node.resource.table_schema is not None:
-        out = visitor.apply_conform(
-            out, node.resource.table_schema, empty_from_schema=True,
-            resource_name=node.resource.name,
-        )
-    return out
+    def _read_and_conform():
+        out = visitor.backend.read_resource(node.resource)
+        if node.resource.table_schema is not None:
+            out = visitor.apply_conform(
+                out, node.resource.table_schema, empty_from_schema=True,
+                resource_name=node.resource.name,
+            )
+        return out
+    return visitor._enrich_native_call(node, RKEY_MOUNTAINASH_REL.READ_RESOURCE, _read_and_conform)
 
 
 def visit_source(node: Any, visitor: Any) -> Any:
     from mountainash.pydata.ingress.pydata_ingress import PydataIngress
     df = PydataIngress.convert(node.data)
-    return visitor.backend.read(df)
+    return visitor._enrich_native_call(
+        node, RKEY_MOUNTAINASH_REL.SOURCE,
+        lambda: visitor.backend.read(df),
+    )
 
 
 def visit_conform(node: Any, visitor: Any) -> Any:
     native = visitor.visit(node.input)
-    return visitor.apply_conform(native, node.spec, contract=node.contract)
+    return visitor._enrich_native_call(
+        node, RKEY_MOUNTAINASH_REL.CONFORM,
+        lambda: visitor.apply_conform(native, node.spec, contract=node.contract),
+    )

src/mountainash/relations/core/relation_system/relation_mapping/registry.py

@@ -50,6 +50,7 @@ class RelationOperationDef:
     options_field: Optional[str] = None    # dict field spread as **kwargs (ExtensionRelNode.options)
     handler: Optional[Callable] = None     # custom compile override: (node, visitor) -> Any
     gate_params: tuple[str, ...] = ()      # extra node fields that gate capability facts
+    wraps_native_call: bool = False        # handler op wraps its native call in _enrich_native_call
 
     def get_signature(self) -> Optional[inspect.Signature]:
         if self.protocol_method is None:

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

@@ -73,15 +73,15 @@ def _iter_function_keys(value: Any, _seen: "set[int] | None" = None):
             yield from _iter_function_keys(item, _seen)
 
 
-def _present_expression_function_keys(node: RelationNode, op: Any) -> frozenset:
-    """Function keys structurally present in *node*'s bound
-    ``EXPRESSION``/``EXPRESSION_LIST`` args -- the caller's evidence for
-    which operation(s) were actually being compiled, used to disambiguate
-    :func:`~mountainash.core.limitations.enrich_materialization` candidates
-    that share a native exception type."""
+def _present_operation_keys(node: RelationNode, op: Any) -> frozenset:
+    """The op's own RKEY plus the function keys structurally present in
+    *node*'s bound ``EXPRESSION``/``EXPRESSION_LIST`` args -- the caller's
+    evidence for which operation(s) were actually being compiled, used to
+    disambiguate :func:`~mountainash.core.limitations.enrich_materialization`
+    candidates that share a native exception type."""
     from mountainash.relations.core.relation_system.relation_mapping.registry import ArgKind
 
-    keys: set = set()
+    keys: set = {op.operation_key}
     for binding in op.args:
         if binding.kind in (ArgKind.EXPRESSION, ArgKind.EXPRESSION_LIST):
             keys.update(_iter_function_keys(getattr(node, binding.field)))
@@ -271,22 +271,36 @@ def _dispatch(self, node: RelationNode, op: Any) -> Any:
         from mountainash.core.limitations import enrich_materialization
 
         if op.handler is not None:
-            # No generic expression-field introspection for handler-routed
-            # ops today -- an empty (non-None) preferred set is
-            # authoritative in enrich_materialization, so this is a
-            # verified no-op unless/until a handler op registers a
-            # MATERIALIZE_RESIDUE fact.
-            return enrich_materialization(
-                self.backend, lambda: op.handler(node, self),
-                prefer_operation_keys=frozenset(),
-            )
+            # Handler ops wrap their own native call (see handlers.py) --
+            # never wrap the whole handler, or a child read/coercion
+            # TypeError would be narrowed under the parent's RKEY.
+            return op.handler(node, self)
         method = getattr(self.backend, op.protocol_method.__name__)
         args = [self._bind(node, b) for b in op.args]  # children compiled OUTSIDE the wrap
         kwargs = self._bind_options(node, op)
-        prefer = _present_expression_function_keys(node, op)
+        prefer = _present_operation_keys(node, op)
+        d = self._authoritative_dialect(node, op)
+        dialect = getattr(self.backend, "dialect", None) if d is _UNRESOLVED else d
         return enrich_materialization(
             self.backend, lambda: method(*args, **kwargs),
             prefer_operation_keys=prefer,
+            dialect=dialect,
+        )
+
+    def _enrich_native_call(self, node: RelationNode, operation_key: Any, fn: Callable[[], Any]) -> Any:
+        """Wrap a single native backend call in residue enrichment, scoped to
+        ``operation_key`` and the authoritative input dialect (item 95)."""
+        from mountainash.core.limitations import enrich_materialization
+        from mountainash.relations.core.relation_system.relation_mapping.registry import (
+            RelationOperationRegistry,
+        )
+        op = RelationOperationRegistry.get(operation_key)
+        d = self._authoritative_dialect(node, op)   # item 95: _UNRESOLVED | str | None
+        dialect = getattr(self.backend, "dialect", None) if d is _UNRESOLVED else d
+        return enrich_materialization(
+            self.backend, fn,
+            prefer_operation_keys=frozenset({operation_key}),
+            dialect=dialect,
         )
 
     def _bind(self, node: RelationNode, binding: Any) -> Any: