diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5013262f..ff23f928 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,9 @@
 
 ## Unreleased — 2026-10-16
 
+### Added
+- Canonical `FieldSpec.dtype` declarations and tagged Frictionless descriptor round trips (333), including dtype-only declarations, native override precedence with mismatch/unverified warnings, and shared conform/inference/schema lowering. Native objects retain parameters in owned in-memory snapshots and are recursively rejected at the descriptor boundary. Integration with native-first schema inspection retains detailed dtype projection and executable native evidence, including nested fields. Existing temporal default-unit loss (334) and NumPy-backed pandas nullable integer casting (252) remain unresolved.
+
 ### Fixed
 - Resource schema inspection (311) uses mountainash-files discovery before execution fallback for remote paths, archives, globs and non-native formats/dialects. Source schema evidence feeds the existing backend compiler and conform path; complete discovery avoids result collection even with execution permission. Bounded observations remain best effort, and missing-source/access/parse errors are not execution retries.
 - Polars 2 migration (332): replace removed String-to-temporal casts with native planning-time parsing on Polars and Narwhals-on-Polars, including nested List/Array/Struct values. Preserve target units, naive wall-clock versus zoned instant semantics, parent nulls, failure policies and window composition. Declared Date ingress and date-range validation now complete through the repaired cast path. The dependency requires `polars>=2,<3`.
diff --git a/docs/website/features-technical/typespec-conform.md b/docs/website/features-technical/typespec-conform.md
index 42beaf8b..922377af 100644
--- a/docs/website/features-technical/typespec-conform.md
+++ b/docs/website/features-technical/typespec-conform.md
@@ -1,6 +1,6 @@
 # TypeSpec and conformance
 
-A schema is a contract about shape. `TypeSpec` is mountainash's serialisable representation of that contract. `conform()` is a single relation method that turns the contract into a transformation: cast types, rename columns, fill nulls, handle missing fields, on any backend.
+A schema is a contract about shape. `TypeSpec` represents that contract, and `conform()` applies its casts, renames, null handling and missing-field policies. Portable declarations serialize to Frictionless descriptors; backend-native dtype objects are supported only in memory.
 
 ## TypeSpec
 
@@ -30,6 +30,49 @@ Under the hood this is a Pydantic model with a list of `FieldSpec` entries — n
 
 `TypeSpec` and `FieldSpec` are deliberately **structurally Frictionless** — flat fields matching the Frictionless layout, not nested custom submodels. If Frictionless gains a property, we add it as a peer field. If we add a property they don't have, it's clearly namespaced.
 
+### Precise dtype declarations
+
+Use a canonical dtype when a broad Frictionless type such as `integer` is insufficient:
+
+```python
+from mountainash.typespec import FieldSpec, TypeSpec, UniversalType
+
+short = FieldSpec(name="id", dtype=ma.MountainashDtype.I8)
+explicit = FieldSpec(name="id", type=UniversalType.INTEGER, dtype=ma.MountainashDtype.I8)
+spec = TypeSpec(fields=[short])
+```
+
+Both forms declare an 8-bit signed integer. Broad-only declarations keep their existing behavior. An incompatible pair, such as `STRING` with `I8`, raises `InvalidDtypeDeclaration`.
+
+Canonical enums persist as a field extension:
+
+```json
+{"name": "id", "type": "integer", "x-mountainash": {
+  "dtype": {"kind": "canonical", "name": "i8"}
+}}
+```
+
+Decimal declarations retain their `kind: "decimal"` precision/scale encoding. Dtype-only `JSON` and native `DURATION` retain broad `ANY` semantics: JSON scalars and arrays do not become GeoJSON, and native durations do not become lexical XSD duration strings. Explicit semantic declarations still apply their parsing rules.
+
+### Native overrides
+
+`backend_type` takes precedence over `dtype`, including decimal declarations:
+
+```python
+import polars as pl
+
+field = FieldSpec(name="id", dtype=ma.MountainashDtype.I8, backend_type="Int64")
+timestamp = FieldSpec(name="at", backend_type=pl.Datetime("ns", "UTC"))
+```
+
+The first field produces Int64 on Polars and emits a `UserWarning` because its canonical declaration is I8. Overrides also warn when equivalence cannot be verified, including parameterized temporal and semantic carriers. Proven equivalent representations do not warn. Invalid strings and objects belonging to a different target raise; they do not fall back to the canonical declaration.
+
+Native strings remain serializable and are interpreted by the selected backend. Native objects retain their parameters in owned conform/validation snapshots but cannot be serialized, including when nested in an object or array-item declaration. Use a native string for a persisted override. Native fingerprints are runtime-local identities, with no stability guarantee across dependency versions.
+
+Schema inspection uses native metadata first and retains detailed dtype parameters. SQLite reports its physical integer/real affinities even when a precise declaration produces narrower typed Polars/Arrow egress. AST fallback inference uses the effective target when known; an unresolved native override reports `UNKNOWN`. `evolve` retains the source's detailed identity; `freeze` compares native parameters when source evidence is available.
+
+Bare canonical timestamp/duration casts retain the existing backend unit defaults. They do not promise nanosecond fidelity; backlog 334 tracks inherited sub-microsecond loss. Use an explicit native target with the required unit when that precision is needed.
+
 ## Sources
 
 `TypeSpec.from_*` and `ma.typespec(...)` accept many inputs:
@@ -96,7 +139,7 @@ df = df[["id", "name", "amount", "ts", "region"]]
 
 ### Cross-backend
 
-`conform()` is cross-backend automatic. The only known limitation today is Ibis coalesce type strictness when `null_fill` mixes string columns with numeric literals — a tracked divergence, not a silent fault.
+`conform()` selects the source's execution backend. Schema conversion is separate from runtime capability: Polars, pandas, Arrow and Ibis have schema converters, while Narwhals uses the existing runtime registry. Null-on-failure casts remain unsupported on Narwhals-routed identities and Ibis-SQLite; decimal limits are listed below. NumPy-backed pandas float/NaN to integer conversion still fails through Narwhals (backlog 252). This pre-existing limitation remains separately tracked and is not repaired by canonical declarations. Ibis also requires compatible types when `null_fill` coalesces values.
 
 ### Structured (`array`/`object`) fields
 
diff --git a/src/mountainash/conform/expressions.py b/src/mountainash/conform/expressions.py
index f7f127cc..fac0e33f 100644
--- a/src/mountainash/conform/expressions.py
+++ b/src/mountainash/conform/expressions.py
@@ -99,6 +99,20 @@ class ConformResult:
     )
 
 
+def _is_null_shape(shape: SourceShape | None) -> bool:
+    from mountainash.core.dtypes import registry
+    from mountainash.typespec.converters import _native_types_equal
+
+    if shape is None or shape.canonical_type is not None or shape.native_dtype is None:
+        return False
+    native = shape.native_dtype
+    for name in ("Null", "null"):
+        null = registry.parse_type_string(name, native.target)
+        if null is not None and _native_types_equal(native.value, null, native.target):
+            return True
+    return False
+
+
 def _structured_carrier(
     field: "FieldSpec", source_shape: SourceShape | None
 ) -> StructuredCarrier | None:
@@ -116,6 +130,10 @@ def _structured_carrier(
     )
     if canonical is native:
         return StructuredCarrier.NATIVE
+    if _is_null_shape(source_shape) and (
+        field.dtype is not None or field.backend_type or field.object_fields or field.item_object_fields
+    ):
+        return StructuredCarrier.NATIVE
     if canonical in {MountainashDtype.STRING, MountainashDtype.JSON}:
         return StructuredCarrier.JSON_TEXT
     if canonical is None:
@@ -209,6 +227,8 @@ def _resolve_declared_type(fld: "FieldSpec", source_name: str) -> DeclaredType:
     from mountainash.typespec.converters import resolve_field_canonical
     from mountainash.typespec.universal_types import UniversalType, to_canonical
 
+    if fld.dtype is not None:
+        return resolve_field_canonical(fld) or UNDETERMINED
     if fld.type == UniversalType.LIST:
         return to_canonical(UniversalType.LIST)  # type: ignore[return-value]
     if fld.categories is not None:
@@ -254,6 +274,10 @@ def _shape_for(
 
 
 def _shape_detail(shape: SourceShape | None) -> str | None:
+    if shape is not None and shape.native_dtype is not None:
+        return str(shape.native_dtype.value)
+    if shape is not None and shape.decimal is not None:
+        return str(shape.decimal)
     if shape is None or shape.canonical_type is None:
         return None
     if shape.canonical_type.name == "LIST" and shape.item_shape is not None:
@@ -274,6 +298,14 @@ def _shape_diff(
 ) -> bool:
     if expected is None or actual is None:
         return False
+    if expected.native_dtype is not None and actual.native_dtype is not None:
+        from mountainash.typespec.converters import _native_types_equal
+        if expected.native_dtype.target is actual.native_dtype.target:
+            return not _native_types_equal(
+                expected.native_dtype.value, actual.native_dtype.value, expected.native_dtype.target,
+            )
+    if expected.decimal is not None:
+        return expected.decimal != actual.decimal
     if expected.canonical_type is None:
         return False
     if actual.canonical_type is None:
@@ -314,12 +346,24 @@ def _shape_diff(
     return False
 
 
-def _expected_shape(fld: "FieldSpec") -> SourceShape | None:
-    from mountainash.core.dtypes import MountainashDtype
-    from mountainash.typespec.converters import resolve_field_canonical
+def _expected_shape(fld: "FieldSpec", type_target: Any = None) -> SourceShape | None:
+    from mountainash.core.dtypes import DecimalDtype, MountainashDtype, NativeDtype, registry
+    from mountainash.typespec.converters import resolve_field_canonical, _resolve_field_target
     from mountainash.typespec.universal_types import parse_universal, to_canonical, UniversalType
 
-    canonical = resolve_field_canonical(fld)
+    canonical = (
+        _resolve_field_target(fld, type_target, emit_warning=False)
+        if type_target is not None else resolve_field_canonical(fld)
+    )
+    if isinstance(canonical, NativeDtype):
+        resolved = registry.from_native(canonical.value, target=canonical.target)
+        return SourceShape(
+            None if isinstance(resolved, DecimalDtype) else resolved,
+            decimal=resolved if isinstance(resolved, DecimalDtype) else None,
+            native_dtype=canonical,
+        )
+    if isinstance(canonical, DecimalDtype):
+        return SourceShape(None, decimal=canonical)
     if canonical is None:
         return None
     if fld.type is UniversalType.GEOPOINT:
@@ -343,7 +387,7 @@ def _expected_shape(fld: "FieldSpec") -> SourceShape | None:
                 SourceShape(
                     MountainashDtype.STRUCT,
                     struct_fields=tuple(
-                        (inner.name, _expected_shape(inner) or SourceShape(None))
+                        (inner.name, _expected_shape(inner, type_target) or SourceShape(None))
                         for inner in fld.item_object_fields
                     ),
                 ),
@@ -353,7 +397,7 @@ def _expected_shape(fld: "FieldSpec") -> SourceShape | None:
         return SourceShape(
             canonical,
             struct_fields=tuple(
-                (inner.name, _expected_shape(inner) or SourceShape(None))
+                (inner.name, _expected_shape(inner, type_target) or SourceShape(None))
                 for inner in fld.object_fields
             ),
         )
@@ -373,12 +417,15 @@ def resolve_conform_output(
     key_resource_name: str | None = None,
     schema_of: Callable[[str], Mapping[str, Any]] | None = None,
     apply_value_transforms: bool = True,
+    type_target: Any = None,
 ) -> ConformOutputContract:
     """Resolve structure, source evidence, drift, and per-field action policy."""
     from mountainash.conform.drift import ColumnDrift, ConformDrift, KeyDrift, TypeDrift
     from mountainash.core.dtypes import CastSafety, DecimalDtype, MountainashDtype, classify_cast
     from mountainash.core.dtypes.canonical import SchemaTypeStatus
     from mountainash.typespec.universal_types import UniversalType
+    from mountainash.core.dtypes import NativeDtype
+    from mountainash.typespec.converters import _resolve_field_target, _native_types_equal
 
     fields_match = spec.fields_match
     if fields_match not in _VALID_FIELDS_MATCH:
@@ -422,15 +469,20 @@ def resolve_conform_output(
     renamed_sources: set[str] = set()
     for field in spec.fields:
         source_name = field.source_name
+        effective = (
+            _resolve_field_target(field, type_target, emit_warning=raise_on_freeze)
+            if type_target is not None
+            else None if field.backend_type else field.dtype or _declared_canonical(field)
+        )
         if available_set is not None and _source_root(source_name) not in available_set:
             if contract.missing_columns == "null_fill":
                 declared = _declared_canonical(field)
-                emitted.append(EmittedField(field, source_name, declared or UNDETERMINED, False, "null_fill", source_shape=SourceShape(None)))
+                emitted.append(EmittedField(field, source_name, declared or UNDETERMINED, False, "null_fill", effective_type=effective, source_shape=SourceShape(None)))
             continue
         renamed = "." not in source_name and source_name != field.name
         if renamed:
             renamed_sources.add(source_name)
-        emitted.append(EmittedField(field, source_name, _resolve_declared_type(field, source_name), renamed, source_shape=_shape_for(actual_shapes, source_name)))  # type: ignore[arg-type]
+        emitted.append(EmittedField(field, source_name, _resolve_declared_type(field, source_name), renamed, effective_type=effective, source_shape=_shape_for(actual_shapes, source_name)))
 
     type_mismatches: list[TypeDrift] = []
     resolved: list[EmittedField] = []
@@ -478,10 +530,23 @@ def resolve_conform_output(
         if "." in em.source_name and actual_shape is None:
             resolved.append(em)
             continue
-        declared = _declared_canonical(em.field)
+        declared = em.effective_type if isinstance(em.effective_type, MountainashDtype) else None
         actual_dtype = (actual_dtypes or {}).get(em.source_name)
         mismatch: TypeDrift | None = None
-        if em.field.dtype is not None:
+        if isinstance(em.effective_type, NativeDtype):
+            actual_native = actual_shape.native_dtype if actual_shape is not None else None
+            comparable = actual_native is not None and actual_native.target is em.effective_type.target
+            if not comparable or not _native_types_equal(
+                actual_native.value, em.effective_type.value, em.effective_type.target,
+            ):
+                mismatch = TypeDrift(
+                    em.field.name, em.effective_type, actual_dtype,
+                    "unsafe" if comparable else "unknown", None,
+                    "representation" if comparable else "unknown",
+                    str(actual_native.value) if actual_native is not None else None,
+                    str(em.effective_type.value), apply_value_transforms,
+                )
+        elif isinstance(em.effective_type, DecimalDtype):
             # Exact decimals match only equal precision/scale. All other source types
             # are mismatches resolved by the existing data_type policy below.
             if actual_shapes is not None or actual_dtypes is not None:
@@ -490,7 +555,7 @@ def resolve_conform_output(
                     else actual_dtype if isinstance(actual_dtype, DecimalDtype)
                     else None
                 )
-                if actual_decimal != em.field.dtype:
+                if actual_decimal != em.effective_type:
                     if actual_decimal is not None:
                         actual_report: Any = actual_decimal
                     elif actual_shape is not None and actual_shape.canonical_type is not None:
@@ -501,18 +566,18 @@ def resolve_conform_output(
                         actual_report = None
                     mismatch = TypeDrift(
                         em.field.name,
-                        em.field.dtype,
+                        em.effective_type,
                         actual_report,
                         "unknown" if actual_report is None else "unsafe",
                         None,
                         "unknown" if actual_report is None else "representation",
                         None if actual_report is None else str(actual_report),
-                        str(em.field.dtype),
+                        str(em.effective_type),
                         apply_value_transforms,
                     )
             declared = None  # skip the canonical comparisons below
         if declared is not None and actual_shapes is not None:
-            expected_shape = _expected_shape(em.field)
+            expected_shape = _expected_shape(em.field, type_target)
             requirement = _shape_detail(expected_shape) or str(declared)
             if actual_shape is None or actual_shape.canonical_type is None:
                 mismatch = TypeDrift(
@@ -577,7 +642,7 @@ def resolve_conform_output(
                     apply_value_transforms,
                 )
         elif declared is not None and actual_dtypes is not None:
-            requirement = _shape_detail(_expected_shape(em.field)) or str(declared)
+            requirement = _shape_detail(_expected_shape(em.field, type_target)) or str(declared)
             if actual_dtype is None or isinstance(actual_dtype, SchemaTypeStatus):
                 mismatch = TypeDrift(
                     em.field.name,
@@ -614,7 +679,7 @@ def resolve_conform_output(
         mismatch = dataclasses.replace(mismatch, action=reported_action, applied=applied)
         type_mismatches.append(mismatch)
         if (
-            em.field.dtype is not None
+            isinstance(em.effective_type, DecimalDtype)
             and em.field.rounding is None
             and apply_value_transforms
             and action in {"coerce", "discard_value", "discard_row"}
@@ -623,7 +688,7 @@ def resolve_conform_output(
             # Unknown inference evidence does not prove conversion is required.
             # Concrete mismatches and terminal resolution still require rounding.
             raise DecimalConversionRequiredError(
-                em.field.name, em.field.dtype, mismatch.actual or "an unknown type"
+                em.field.name, em.effective_type, mismatch.actual or "an unknown type"
             )
         if action == "evolve":
             em = dataclasses.replace(em, type_action="evolve", effective_type=mismatch.actual)
@@ -696,22 +761,23 @@ def _build_field_expr(
     type_action: str = "coerce",
     declared_type: Optional[DeclaredType] = None,
     source_shape: SourceShape | None = None,
+    effective_type: Any = None,
 ) -> FieldBuildResult:
     """Lower one field through the backend-neutral expression API."""
     import mountainash as ma
-    from mountainash.core.dtypes import MountainashDtype
+    from mountainash.core.dtypes import DecimalDtype, MountainashDtype, NativeDtype
     from mountainash.expressions.core.expression_protocols.api_builders.substrait.prtcl_api_bldr_cast import CaseFailureBehaviour
     from mountainash.typespec.spec import FieldSpec
     from mountainash.typespec.universal_types import UniversalType, to_canonical
 
     fld = field
     if type_action == "null_fill":
-        if fld.dtype is not None:
-            # Declared exact decimal: a typed null keeps the declared precision/scale.
-            return FieldBuildResult(ma.lit(None, dtype=fld.dtype).name.alias(fld.name))
         output = ma.lit(None)
-        if isinstance(declared_type, MountainashDtype):
-            output = output.cast(declared_type)
+        target = effective_type if effective_type is not None else declared_type
+        if isinstance(target, DecimalDtype):
+            output = ma.lit(None, dtype=target)
+        elif isinstance(target, (MountainashDtype, NativeDtype)):
+            output = output.cast(target)
         return FieldBuildResult(output.name.alias(fld.name))
 
     is_dotted = "." in source_name
@@ -722,9 +788,9 @@ def _build_field_expr(
     else:
         expr = ma.col(source_name)
 
-    if fld.dtype is not None and (
+    if isinstance(effective_type, DecimalDtype) and (
         source_shape is None
-        or source_shape.decimal == fld.dtype
+        or source_shape.decimal == effective_type
         or type_action == "evolve"
     ):
         # Matching decimals and metadata-free ingress keep their existing pass-through.
@@ -733,6 +799,17 @@ def _build_field_expr(
     shape_known = source_shape is not None
     shape = source_shape or SourceShape(None)
     canonical = shape.canonical_type
+    null_container_target = None
+    if (
+        fld.type in {UniversalType.ARRAY, UniversalType.OBJECT}
+        and (fld.dtype is not None or fld.backend_type or fld.object_fields or fld.item_object_fields)
+        and _is_null_shape(shape) and type_action != "evolve"
+    ):
+        from mountainash.typespec.converters import _resolve_field_native
+        target = shape.native_dtype.target
+        null_container_target = effective_type if isinstance(effective_type, NativeDtype) else NativeDtype(
+            value=_resolve_field_native(fld, target), target=target,
+        )
     if _structured_carrier(fld, shape) in {
         StructuredCarrier.JSON_TEXT,
         StructuredCarrier.OPAQUE,
@@ -763,6 +840,7 @@ def _build_field_expr(
     if (
         shape_known
         and canonical is None
+        and null_container_target is None
         and (
             (fld.type in typed_shapes and not unknown_lexical)
             or unknown_temporal
@@ -774,7 +852,7 @@ def _build_field_expr(
             requirement="source shape for typed operation",
         )
     incompatible_source = False
-    if shape_known:
+    if shape_known and null_container_target is None:
         allowed = {
             UniversalType.LIST: {MountainashDtype.STRING},
             UniversalType.ARRAY: {MountainashDtype.LIST},
@@ -890,13 +968,13 @@ def _build_field_expr(
         UniversalType.GEOPOINT,
         UniversalType.GEOJSON,
     }
-    lexical = (
+    lexical = null_container_target is None and (
         not shape_known
         or canonical is None
         or canonical is MountainashDtype.STRING
         or (fld.type is UniversalType.GEOJSON and canonical is MountainashDtype.JSON)
     )
-    if fld.dtype is not None and shape.decimal is not None:
+    if isinstance(effective_type, DecimalDtype) and shape.decimal is not None:
         lexical = False
     sentinels = fld.missing_values if fld.missing_values is not None else schema_missing_values
     from mountainash.typespec._categorical import categorical_values
@@ -921,7 +999,7 @@ def _build_field_expr(
     transform_input = expr
     if fld.null_fill is not None:
         fill = ma.lit(fld.null_fill)
-        if fld.dtype is not None and shape.decimal is not None:
+        if isinstance(effective_type, DecimalDtype) and shape.decimal is not None:
             # Preserve fill digits: casting to the source scale first can double-round.
             expr = expr.cast(MountainashDtype.STRING)
             fill = fill.cast(MountainashDtype.STRING)
@@ -995,8 +1073,10 @@ def _placeholder(inner):
     residue: list[MaterializationResidueCheck] = []
     if type_action == "evolve":
         return FieldBuildResult(transform_input.name.alias(fld.name))
-    if fld.dtype is not None:
-        expr = expr.cast(fld.dtype, rounding=fld.rounding, failure_behavior=failure)
+    if null_container_target is not None:
+        expr = expr.cast(null_container_target, failure_behavior=failure)
+    elif isinstance(effective_type, DecimalDtype):
+        expr = expr.cast(effective_type, rounding=fld.rounding, failure_behavior=failure)
     elif fld.type == UniversalType.LIST and lexical:
         expr = expr.str.parse_list(
             item_type=fld.item_type or "string",
@@ -1004,6 +1084,13 @@ def _placeholder(inner):
             field_name=fld.name,
             failure_behavior=failure,
         )
+    elif (
+        isinstance(effective_type, NativeDtype)
+        and fld.type in {UniversalType.ARRAY, UniversalType.OBJECT}
+        and not lexical
+    ):
+        # The native container replaces its declared child representation.
+        pass
     elif (
         fld.type == UniversalType.ARRAY
         and canonical is MountainashDtype.LIST
@@ -1037,7 +1124,7 @@ def _placeholder(inner):
         # Plain native containers have no child schema to apply.  The source
         # already has the required physical representation; preserve it.
         pass
-    elif fld.categories is not None:
+    elif fld.categories is not None and fld.dtype is None and not isinstance(effective_type, NativeDtype):
         from mountainash.typespec._categorical import categorical_values
         expr = expr.cat.cast(
             value_type=(fld.type.value if fld.type and fld.type is not UniversalType.ANY else "string"),
@@ -1094,10 +1181,18 @@ def _placeholder(inner):
         expr = method(fld.format, field_name=fld.name, failure_behavior=failure)
     elif fld.type == UniversalType.DATETIME and fld.format == "default" and lexical:
         expr = expr.dt.parse_default(field_name=fld.name, failure_behavior=failure)
-    elif fld.type and fld.type is not UniversalType.ANY:
+    elif fld.type and fld.type is not UniversalType.ANY and fld.dtype is None and not isinstance(effective_type, NativeDtype):
         target = to_canonical(fld.type)
         if target is not None:
             expr = expr.cast(target, failure_behavior=failure)
+    if null_container_target is None and (
+        isinstance(effective_type, NativeDtype)
+        or (
+            fld.dtype is not None and isinstance(effective_type, MountainashDtype)
+            and effective_type not in {MountainashDtype.LIST, MountainashDtype.STRUCT}
+        )
+    ):
+        expr = expr.cast(effective_type, failure_behavior=failure)
 
     output = expr.name.alias(fld.name)
     discard = None
@@ -1118,6 +1213,7 @@ def _build_conform_exprs(
     key_resource_name: str | None = None,
     schema_of: Callable[[str], Mapping[str, Any]] | None = None,
     apply_value_transforms: bool = True,
+    type_target: Any = None,
     raise_on_freeze: bool = True,
 ) -> ConformResult:
     schema_missing_values = spec.missing_values or []
@@ -1133,6 +1229,7 @@ def _build_conform_exprs(
         key_resource_name=key_resource_name,
         schema_of=schema_of,
         apply_value_transforms=apply_value_transforms,
+        type_target=type_target,
         raise_on_freeze=raise_on_freeze,
     )
     structured_field_plans = _build_structured_field_plans(
@@ -1146,14 +1243,16 @@ def _build_conform_exprs(
     exprs: list[Any] = []
     row_filters: list[Any] = []
     residue_checks: list[MaterializationResidueCheck] = []
-    from mountainash.core.dtypes import MountainashDtype
+    from mountainash.core.dtypes import DecimalDtype, MountainashDtype, NativeDtype
     for emitted in output_contract.emitted:
         if not apply_value_transforms:
             import mountainash as ma
             if emitted.type_action == "null_fill":
-                source = ma.lit(None, dtype=emitted.field.dtype) if emitted.field.dtype is not None else ma.lit(None)
-                if emitted.field.dtype is None and isinstance(emitted.declared_type, MountainashDtype):
-                    source = source.cast(emitted.declared_type)
+                source = ma.lit(None)
+                if isinstance(emitted.effective_type, DecimalDtype):
+                    source = ma.lit(None, dtype=emitted.effective_type)
+                elif isinstance(emitted.effective_type, (MountainashDtype, NativeDtype)):
+                    source = source.cast(emitted.effective_type)
                 exprs.append(source.name.alias(emitted.field.name))
                 continue
             source = ma.col(emitted.source_name.split(".", 1)[0])
@@ -1175,6 +1274,7 @@ def _build_conform_exprs(
             type_action=emitted.type_action,
             declared_type=emitted.declared_type,
             source_shape=emitted.source_shape,
+            effective_type=emitted.effective_type,
         )
         exprs.append(built.output)
         if built.discard_row_filter is not None:
diff --git a/src/mountainash/core/dtypes/canonical.py b/src/mountainash/core/dtypes/canonical.py
index c3f67325..3020671d 100644
--- a/src/mountainash/core/dtypes/canonical.py
+++ b/src/mountainash/core/dtypes/canonical.py
@@ -357,7 +357,7 @@ def parse_cast_target(
     unchanged. Native backend dtype objects wrap in NativeDtype WITHOUT
     normalization (parameters preserved).
     """
-    if isinstance(value, DecimalDtype):
+    if isinstance(value, (DecimalDtype, NativeDtype)):
         return value
     try:
         return parse_dtype(value)
diff --git a/src/mountainash/pydata/ingress/custom_type_helpers.py b/src/mountainash/pydata/ingress/custom_type_helpers.py
index 5f30c6c5..6dd9bbc5 100644
--- a/src/mountainash/pydata/ingress/custom_type_helpers.py
+++ b/src/mountainash/pydata/ingress/custom_type_helpers.py
@@ -158,10 +158,9 @@ def apply_native_conversions_to_dataframe(
     """
     Apply native type conversions to a DataFrame using vectorized operations.
 
-    This is TIER 1 (DataFrame layer) of the hybrid strategy:
-    - Apply ONLY native conversions (string→int, float→bool)
-    - Use vectorized operations (MUCH FASTER - 12x!)
-    - Apply other native operations (rename, null_fill) via Relation.conform()
+    TIER 1 uses the shared conform expressions for native casting, rename and
+    null filling. Target resolution precedes execution so invalid overrides
+    cannot be swallowed by the existing best-effort execution policy.
 
     Args:
         df: Polars DataFrame
@@ -171,63 +170,27 @@ def apply_native_conversions_to_dataframe(
     Returns:
         DataFrame with native conversions applied
     """
-    from mountainash.core.lazy_imports import import_polars
-    from mountainash.core.dtypes import TypeTarget, registry
-    from mountainash.typespec.converters import resolve_field_canonical
-    from mountainash.typespec.universal_types import UniversalType
-
-    pl = import_polars()
-    if pl is None:
-        raise ImportError("polars is required for this operation")
 
     if not native_conversions:
         return df
 
-    # Build cast dictionary for vectorized casting
-    cast_dict = {}
-    for col_name, field_spec in native_conversions.items():
-        # Only cast if there's a non-ANY type declared and column exists
-        if field_spec.type and field_spec.type != UniversalType.ANY and col_name in df.columns:
-            canon = resolve_field_canonical(field_spec)
-            if canon is not None:
-                cast_dict[col_name] = registry.to_native_schema(canon, TypeTarget.POLARS)
-
-    # Apply vectorized cast (MUCH FASTER than element-wise!)
-    if cast_dict:
-        try:
-            df = df.cast(cast_dict)
-            logger.info(
-                f"Applied native conversions (vectorized) to {len(cast_dict)} columns"
-            )
-        except Exception as e:
-            logger.warning(
-                f"Vectorized cast failed: {e}. "
-                "Falling back to column-by-column casting."
-            )
-            for col_name, dtype in cast_dict.items():
-                try:
-                    df = df.with_columns(pl.col(col_name).cast(dtype))
-                except Exception as col_error:
-                    logger.warning(
-                        f"Cast failed for column '{col_name}': {col_error}. "
-                        "Keeping original type."
-                    )
-
-    # Apply other native operations (rename, null_fill) via expressions
     import mountainash as ma
     from mountainash.typespec.spec import TypeSpec
     from mountainash.conform.expressions import _build_conform_exprs
+    from mountainash.core.dtypes import TypeTarget
+    from mountainash.typespec.source_shape import extract_source_shapes
 
-    # Permissive application (see the STEP 5 native path below): rename/null_fill
-    # expressions are applied via with_columns without available_columns, so the
-    # spec must use "open" (the model default is now the strict "exact").
+    # Preserve columns outside the requested native conversions.
     native_spec = TypeSpec(
         fields=[f for f in native_conversions.values()],
         fields_match="open",
     )
+    conform_result = _build_conform_exprs(
+        native_spec, available_columns=df.columns, actual_shapes=extract_source_shapes(df),
+        type_target=TypeTarget.POLARS,
+    )
     try:
-        exprs = _build_conform_exprs(native_spec)
-        df = ma.relation(df).with_columns(*exprs).to_polars()
+        df = ma.relation(df).with_columns(*conform_result.exprs).to_polars()
     except Exception as e:
         logger.warning(
             f"Error applying native operations via expressions: {e}. "
@@ -413,26 +376,7 @@ def apply_hybrid_conversion(
 
     # STEP 5: Apply NATIVE operations via expressions (TIER 1)
     if native:
-        import mountainash as ma
-        from mountainash.typespec.spec import TypeSpec
-        from mountainash.conform.expressions import _build_conform_exprs
-
-        # Native casting applies per-column expressions via with_columns and
-        # must not enforce schema shape — it runs without available_columns,
-        # so it needs the permissive "open" mode explicitly (the model default
-        # is now the strict "exact", which requires column information).
-        native_spec = TypeSpec(
-            fields=[f for f in native.values()],
-            fields_match="open",
-        )
-        try:
-            conform_result = _build_conform_exprs(native_spec)
-            df = ma.relation(df).with_columns(*conform_result.exprs).to_polars()
-        except Exception as e:
-            logger.warning(
-                f"Error applying native operations via expressions: {e}. "
-                "DataFrame may not have all operations applied."
-            )
+        df = apply_native_conversions_to_dataframe(df, native)
 
     return df
 
diff --git a/src/mountainash/relations/core/unified_visitor/relation_visitor.py b/src/mountainash/relations/core/unified_visitor/relation_visitor.py
index 94172dd6..b0fc186c 100644
--- a/src/mountainash/relations/core/unified_visitor/relation_visitor.py
+++ b/src/mountainash/relations/core/unified_visitor/relation_visitor.py
@@ -821,6 +821,8 @@ def apply_conform(
         from mountainash.core.types import BackendCapabilityError
         from mountainash.typespec.source_shape import extract_source_shapes
         from mountainash.relations.schema_inference import _schema_from_dataframe
+        from mountainash.typespec.converters import _execution_type_target
+        type_target = _execution_type_target(self.execution_context.target.identity.family)
 
         # dtype-aware detection first ({} degrades honestly for an
         # unrecognized backend or a genuinely zero-column frame); falls back
@@ -842,7 +844,7 @@ def apply_conform(
         # Validate before resolve_contract(): resolve_contract's
         # FIELDS_MATCH_PRESETS[fields_match] lookup deliberately raises
         try:
-            actual_shapes = extract_source_shapes(native)
+            actual_shapes = extract_source_shapes(native, target=type_target)
         except (TypeError, ImportError):
             actual_shapes = None
         # bare KeyError on an unknown preset name (pinned by
@@ -908,6 +910,7 @@ def apply_conform(
             key_resource_name=key_resource_name,
             schema_of=key_schema_of,
             apply_value_transforms=apply_value_transforms,
+            type_target=type_target,
             raise_on_freeze=not schema_only,
         )
         if schema_only:
diff --git a/src/mountainash/relations/schema_inference.py b/src/mountainash/relations/schema_inference.py
index d10df0e7..a10ba314 100644
--- a/src/mountainash/relations/schema_inference.py
+++ b/src/mountainash/relations/schema_inference.py
@@ -35,9 +35,7 @@
 from mountainash.expressions.core.output_names import resolve_output_names
 from mountainash.relations.core.projection_names import require_projection_names
 from mountainash.typespec.frictionless import typespec_from_frictionless
-from mountainash.typespec.spec import FieldSpec
 from mountainash.typespec.converters import resolve_field_canonical
-from mountainash.typespec.universal_types import parse_universal
 from mountainash.core.types import (
     is_narwhals_dataframe,
     is_narwhals_lazyframe,
@@ -307,30 +305,21 @@ def _schema_from_table_schema(
     table_schema: dict,
 ) -> dict[str, _DetailedDtype | SchemaTypeStatus]:
     """Extract field-aware canonical schema from a Frictionless table schema."""
-    fields = table_schema.get("fields", [])
-    if not fields:
-        return {}
     result: dict[str, _DetailedDtype | SchemaTypeStatus] = {}
-    for raw_field in fields:
+    for raw_field in table_schema.get("fields", []):
         name = raw_field.get("name")
-        if not name or not raw_field.get("type"):
-            if name:
-                result[name] = SchemaTypeStatus.UNKNOWN
+        if not name:
             continue
         try:
-            field = FieldSpec(
-                name=name,
-                type=parse_universal(raw_field["type"]),
-                format=raw_field.get("format", "default"),
-                item_type=raw_field.get("itemType"),
-                delimiter=raw_field.get("delimiter"),
-            )
+            field = typespec_from_frictionless({"fields": [raw_field]}).fields[0]
             canonical = resolve_field_canonical(field)
         except (KeyError, TypeError, ValueError):
             result[name] = SchemaTypeStatus.UNKNOWN
             continue
         result[name] = (
-            SchemaTypeStatus.UNCONSTRAINED if canonical is None else canonical
+            SchemaTypeStatus.UNKNOWN if field.backend_type
+            else SchemaTypeStatus.UNCONSTRAINED if canonical is None and raw_field.get("type")
+            else SchemaTypeStatus.UNKNOWN if canonical is None else canonical
         )
     return result
 
@@ -348,7 +337,8 @@ def _schema_from_typespec(
             result[field.name] = SchemaTypeStatus.UNKNOWN
             continue
         result[field.name] = (
-            SchemaTypeStatus.UNCONSTRAINED if canonical is None else canonical
+            SchemaTypeStatus.UNKNOWN if field.backend_type
+            else SchemaTypeStatus.UNCONSTRAINED if canonical is None else canonical
         )
     return result
 
@@ -460,10 +450,30 @@ def infer_schema(
         )
 
         node_id = f"conform:{len(_drifts)}" if _drifts is not None else None
+        from mountainash.relations.core.execution.location import IdentityTokens, LocationResolver
+        from mountainash.relations.core.errors import UnresolvedExecutionLocationError
+        from mountainash.relations.dag.errors import RelationDAGRequired
+        from mountainash.typespec.converters import _execution_type_target
+        from mountainash.typespec.source_shape import extract_source_shapes
+        try:
+            location = LocationResolver(IdentityTokens(), identity_resolver=identity_resolver).resolve(node)
+            type_target = _execution_type_target(location.family)
+        except UnresolvedExecutionLocationError:
+            raise
+        except (RelationDAGRequired, ValueError):
+            type_target = None
+        actual_shapes = (
+            extract_source_shapes(node.input.dataframe, target=type_target)
+            if isinstance(node.input, ReadRelNode)
+            and type_target is not None
+            and not isinstance(node.input.dataframe, (list, tuple, dict)) else None
+        )
         contract = resolve_conform_output(
             spec,
             available_columns=list(input_schema.keys()),
             actual_dtypes=input_schema,
+            actual_shapes=actual_shapes,
+            type_target=type_target,
             contract=resolved_contract,
             node_identity=(node_id, None, getattr(spec, "name", None)),
             raise_on_freeze=False,
@@ -578,12 +588,9 @@ def _declared_dtype_for_infer(
 ) -> _DetailedDtype | SchemaTypeStatus:
     """Resolve an EmittedField's declared_type against an upstream schema.
 
-    - ``type_action == "evolve"`` (item 48 Task 9, R2) → the output keeps the
-      source's actual type: ``em.effective_type`` when the data_type drift
-      loop resolved a concrete dtype, else ``UNKNOWN`` (evidence was
-      unknowable — honest non-assessment, not a guess). This check runs
-      first because "evolve" overrides whatever ``declared_type`` would
-      otherwise resolve to.
+    - ``type_action == "evolve"`` keeps detailed input-schema evidence where
+      available, then the drift resolver's effective type, else ``UNKNOWN``.
+      This takes precedence over the declared target.
     - Concrete :class:`MountainashDtype` → returned as-is.
     - ``PASSTHROUGH`` → look up ``em.source_name`` in ``input_schema`` (UNKNOWN
       if absent — e.g. dotted struct child where the root is present but the
@@ -592,13 +599,17 @@ def _declared_dtype_for_infer(
       pre-compile; e.g. ANY + null_fill, dotted ANY, non-Polars categorical).
     """
     if em.type_action == "evolve":
-        return (
-            em.effective_type
-            if em.effective_type is not None
-            else SchemaTypeStatus.UNKNOWN
+        return input_schema.get(
+            em.source_name,
+            em.effective_type if em.effective_type is not None else SchemaTypeStatus.UNKNOWN,
         )
-    if em.field.dtype is not None:
-        return em.field.dtype
+    from mountainash.core.dtypes import DecimalDtype, NativeDtype, registry
+    if isinstance(em.effective_type, NativeDtype):
+        return registry.from_native(em.effective_type.value, target=em.effective_type.target) or SchemaTypeStatus.UNKNOWN
+    if em.field.backend_type and em.effective_type is None:
+        return SchemaTypeStatus.UNKNOWN
+    if isinstance(em.effective_type, (MountainashDtype, DecimalDtype)):
+        return em.effective_type
     dt = em.declared_type
     if dt is PASSTHROUGH:
         return input_schema.get(em.source_name, SchemaTypeStatus.UNKNOWN)
diff --git a/src/mountainash/typespec/_fingerprint.py b/src/mountainash/typespec/_fingerprint.py
index 79dba960..10e5a105 100644
--- a/src/mountainash/typespec/_fingerprint.py
+++ b/src/mountainash/typespec/_fingerprint.py
@@ -7,10 +7,16 @@
 import hashlib
 from types import MappingProxyType
 from typing import Any
+from copy import deepcopy
+import pickle
+
+from mountainash.core.dtypes import NativeDtype
 
 
 def freeze_declaration(value: Any) -> Any:
     """Recursively replace mutable declaration values with immutable tagged data."""
+    if isinstance(value, NativeDtype):
+        return deepcopy(value)
     if isinstance(value, Enum):
         return ("__enum__", value.__class__.__module__, value.__class__.__qualname__, value.value)
     if is_dataclass(value) and not isinstance(value, type):
@@ -45,6 +51,12 @@ def freeze_typespec(spec: Any) -> Mapping[str, Any]:
 
 def _canonical_bytes(value: Any) -> bytes:
     """Encode the closed frozen-declaration vocabulary deterministically."""
+    if isinstance(value, NativeDtype):
+        native_class = value.value if isinstance(value.value, type) else type(value.value)
+        return b"native" + _canonical_bytes((
+            value.target.value, native_class.__module__, native_class.__qualname__,
+            pickle.dumps(value.value, protocol=5),
+        ))
     if value is None:
         return b"n"
     if type(value) is bool:  # noqa: E721 — bool has its own canonical tag
diff --git a/src/mountainash/typespec/converters.py b/src/mountainash/typespec/converters.py
index 4cd8f858..d3d1d1d0 100644
--- a/src/mountainash/typespec/converters.py
+++ b/src/mountainash/typespec/converters.py
@@ -12,8 +12,11 @@
 from __future__ import annotations
 
 from typing import TYPE_CHECKING, Any, Dict
+import warnings
 
 from mountainash.core.dtypes import (
+    DecimalDtype,
+    NativeDtype,
     InvalidBackendTypeError,
     MountainashDtype,
     TypeTarget,
@@ -21,6 +24,7 @@
     registry,
 )
 from mountainash.typespec.errors import InvalidGeospatialFormatError
+from mountainash.core.dtypes.errors import DtypeMappingError
 from mountainash.typespec.universal_types import UniversalType, to_canonical
 
 if TYPE_CHECKING:
@@ -54,6 +58,12 @@ def resolve_field_canonical(field: "FieldSpec") -> Any:
     Every other UniversalType delegates to to_canonical(field.type)
     unchanged.
     """
+    broad = _resolve_broad_field_canonical(field)
+    return field.dtype if field.dtype is not None else broad
+
+
+def _resolve_broad_field_canonical(field: "FieldSpec") -> Any:
+    """Resolve authored broad semantics independently of precise refinements."""
     fmt = field.format
     if field.type is UniversalType.GEOPOINT:
         if not isinstance(fmt, str) or fmt not in _GEOPOINT_FORMAT_CANONICAL:
@@ -73,18 +83,81 @@ def resolve_field_canonical(field: "FieldSpec") -> Any:
                 sorted(_GEOJSON_FORMATS),
             )
         return MountainashDtype.JSON
+    if field.type is UniversalType.STRING and field.format == "binary":
+        return MountainashDtype.BINARY
     return to_canonical(field.type)
 
 
 # ============================================================================
 # Shared resolution core
 # ============================================================================
+def _execution_type_target(family: Any) -> TypeTarget | None:
+    """Translate an execution family, not a dataframe carrier, to its dtype domain."""
+    if family is None:
+        return None
+    if family.value in ("pandas", "pyarrow", "narwhals"):
+        return TypeTarget.NARWHALS
+    return TypeTarget(family.value)
+
+
+def _native_types_equal(left: Any, right: Any, target: TypeTarget) -> bool:
+    """Compare native types in their owning domain, retaining parameters."""
+    if target is TypeTarget.IBIS:
+        import ibis.expr.datatypes as dt
+        return dt.dtype(left) == dt.dtype(right)
+    if target is TypeTarget.PANDAS:
+        import pandas as pd
+        return pd.api.types.pandas_dtype(left) == pd.api.types.pandas_dtype(right)
+    return left == right
+
+
+def _resolve_field_target(
+    field: "FieldSpec", target: TypeTarget, *, emit_warning: bool = True,
+) -> MountainashDtype | DecimalDtype | NativeDtype | None:
+    """Resolve a declaration once, honoring its explicit native override."""
+    declared = resolve_field_canonical(field)
+    override = field.backend_type
+    if override is None or (isinstance(override, str) and not override):
+        return declared
+    if isinstance(override, NativeDtype):
+        if override.target is not target:
+            raise DtypeMappingError(
+                f"field {field.name!r}: native override belongs to {override.target}, not {target}"
+            )
+        effective = override
+    else:
+        native = registry.parse_type_string(override, target)
+        if native is None:
+            raise InvalidBackendTypeError(field.name, override, target)
+        effective = NativeDtype(value=native, target=target)
+    if field.dtype is not None and emit_warning:
+        semantic_or_parameterized = declared in (
+            MountainashDtype.JSON, MountainashDtype.XSD_DURATION,
+            MountainashDtype.XSD_YEAR, MountainashDtype.XSD_YEARMONTH,
+            MountainashDtype.TIMESTAMP, MountainashDtype.DURATION,
+            MountainashDtype.LIST, MountainashDtype.STRUCT,
+        )
+        equivalent = False
+        if not semantic_or_parameterized:
+            expected = registry.to_native_schema(declared, target)
+            equivalent = _native_types_equal(expected, effective.value, target)
+        if not equivalent:
+            reason = "equivalence unverified" if semantic_or_parameterized else "dtype mismatch"
+            warnings.warn(
+                f"field {field.name!r} on {target.value}: {reason} between "
+                f"canonical {declared!r} and native override {effective.value!r}; native override wins",
+                UserWarning, stacklevel=3,
+            )
+    return effective
+
+
 
 def _resolve_field_native(field: "FieldSpec", target: TypeTarget) -> Any:
-    """Resolve declared decimals, categories, backend overrides, canonical types, and containers."""
-    if field.dtype is not None:
-        return registry.to_native_schema(field.dtype, target)
-    if field.categories is not None and target in (TypeTarget.POLARS, TypeTarget.PANDAS):
+    """Lower the effective target and enrich canonical container declarations."""
+    canon = _resolve_field_target(field, target)
+    if isinstance(canon, NativeDtype):
+        return canon.value
+    if field.dtype is None and field.categories is not None and target in (TypeTarget.POLARS, TypeTarget.PANDAS):
         from mountainash.typespec._categorical import categorical_values
         values = categorical_values(field.categories)
         if target is TypeTarget.POLARS:
@@ -93,12 +166,6 @@ def _resolve_field_native(field: "FieldSpec", target: TypeTarget) -> Any:
             return pl.Enum([str(v) for v in values]) if field.categories_ordered else pl.Categorical
         import pandas as pd
         return pd.CategoricalDtype(categories=values, ordered=bool(field.categories_ordered))
-    if field.backend_type:
-        parsed = registry.parse_type_string(field.backend_type, target)
-        if parsed is not None:
-            return parsed
-        raise InvalidBackendTypeError(field.name, field.backend_type, target)
-    canon = resolve_field_canonical(field)
     if canon is None:
         canon = MountainashDtype.STRING
     native = registry.to_native_schema(canon, target)
diff --git a/src/mountainash/typespec/extraction.py b/src/mountainash/typespec/extraction.py
index 2b75483b..d6b12db9 100644
--- a/src/mountainash/typespec/extraction.py
+++ b/src/mountainash/typespec/extraction.py
@@ -85,17 +85,20 @@ def _field_from_source_shape(
                 ]
 
     if shape.decimal is not None:
-        # Declared exact decimal: precision/scale are the schema identity, so the
-        # backend-specific type string is redundant (and mutually exclusive).
+        # Exact decimals retain portable precision/scale without a native spelling.
         return FieldSpec(
             name=name,
             type=UniversalType.NUMBER,
             dtype=shape.decimal,
         )
 
+    from mountainash.core.dtypes import MountainashDtype
+    if shape.canonical_type in (MountainashDtype.JSON, MountainashDtype.DURATION):
+        universal_type = UniversalType.ANY
     return FieldSpec(
         name=name,
         type=universal_type,
+        dtype=shape.canonical_type,
         object_fields=object_fields,
         item_object_fields=item_object_fields,
         backend_type=backend_type,
diff --git a/src/mountainash/typespec/frictionless.py b/src/mountainash/typespec/frictionless.py
index a6950e3d..b8c5eeb1 100644
--- a/src/mountainash/typespec/frictionless.py
+++ b/src/mountainash/typespec/frictionless.py
@@ -18,7 +18,7 @@
 
 from mountainash.conform.contract import validate_contract_dict
 
-from mountainash.core.dtypes import DecimalDtype
+from mountainash.core.dtypes import DecimalDtype, MountainashDtype, NativeDtype
 from mountainash.core.dtypes.errors import UnknownDtypeError
 
 from .errors import InvalidDtypeDeclaration, InvalidFieldMatchDeclaration, InvalidKeyShapeError
@@ -248,8 +248,8 @@ def _parse_constraints(
 # Export: TypeSpec → Frictionless dict
 # ---------------------------------------------------------------------------
 
-def _dtype_from_extension(field_name: str, raw: Any) -> Optional[DecimalDtype]:
-    """Rebuild a ``DecimalDtype`` from ``x-mountainash.dtype``; a malformed value raises.
+def _dtype_from_extension(field_name: str, raw: Any) -> Optional[MountainashDtype | DecimalDtype]:
+    """Rebuild a canonical dtype from ``x-mountainash.dtype``; malformed values raise.
 
     The extension lives in our own namespace, so an unrecognised kind or a missing/invalid
     parameter is a malformed declaration, never silently dropped.
@@ -259,9 +259,14 @@ def _dtype_from_extension(field_name: str, raw: Any) -> Optional[DecimalDtype]:
     if not isinstance(raw, Mapping):
         raise InvalidDtypeDeclaration(field_name, "x-mountainash.dtype must be an object", raw)
     kind = raw.get("kind")
+    if kind == "canonical":
+        try:
+            return MountainashDtype(raw.get("name"))
+        except (ValueError, TypeError) as exc:
+            raise InvalidDtypeDeclaration(field_name, "unknown canonical dtype", raw.get("name")) from exc
     if kind != "decimal":
         raise InvalidDtypeDeclaration(
-            field_name, "x-mountainash.dtype kind must be 'decimal'", kind
+            field_name, "x-mountainash.dtype kind must be 'canonical' or 'decimal'", kind
         )
     precision, scale = raw.get("precision"), raw.get("scale")
     try:
@@ -331,8 +336,14 @@ def _field_to_frictionless_dict(fspec: "FieldSpec") -> Dict[str, Any]:
     if fspec.constraints and fspec.constraints.enum_weights is not None:
         field_extensions["enum_weights"] = fspec.constraints.enum_weights
     if fspec.backend_type is not None:
+        if isinstance(fspec.backend_type, NativeDtype):
+            raise InvalidDtypeDeclaration(
+                fspec.name, "native dtype objects are in-memory-only; use a native string for persistence"
+            )
         field_extensions["backend_type"] = fspec.backend_type
-    if fspec.dtype is not None:
+    if isinstance(fspec.dtype, MountainashDtype):
+        field_extensions["dtype"] = {"kind": "canonical", "name": fspec.dtype.value}
+    elif isinstance(fspec.dtype, DecimalDtype):
         field_extensions["dtype"] = {
             "kind": "decimal",
             "precision": fspec.dtype.precision,
diff --git a/src/mountainash/typespec/source_shape.py b/src/mountainash/typespec/source_shape.py
index 2328da2a..e70a6825 100644
--- a/src/mountainash/typespec/source_shape.py
+++ b/src/mountainash/typespec/source_shape.py
@@ -1,10 +1,11 @@
 """Schema-only structural evidence extracted from supported table carriers."""
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field, replace
+from copy import deepcopy
 from typing import Any
 
-from mountainash.core.dtypes import DecimalDtype, MountainashDtype, TypeTarget, registry
+from mountainash.core.dtypes import DecimalDtype, MountainashDtype, NativeDtype, TypeTarget, registry
 from mountainash.core.dtypes.errors import UnknownDtypeError
 
 from mountainash.core.types import (
@@ -27,6 +28,7 @@ class SourceShape:
     item_shape: SourceShape | None = None
     struct_fields: tuple[tuple[str, SourceShape], ...] = ()
     decimal: DecimalDtype | None = None
+    native_dtype: NativeDtype | None = field(default=None, compare=False)
 
     def __post_init__(self) -> None:
         if self.decimal is not None and (
@@ -84,9 +86,36 @@ def _shape_from_dtype(dtype: Any) -> SourceShape:
     return SourceShape(canonical)
 
 
+def _from_native_dtype(native: Any, target: TypeTarget) -> SourceShape:
+    """Project detailed metadata while retaining executable native evidence."""
+    shape = _shape_from_dtype(_resolve(native, target))
+    evidence = NativeDtype(value=deepcopy(native), target=target)
+    if target is TypeTarget.PANDAS and (arrow := getattr(native, "pyarrow_dtype", None)) is not None:
+        return replace(_from_native_dtype(arrow, TypeTarget.PYARROW), native_dtype=evidence)
+    if target is TypeTarget.PYARROW and hasattr(native, "nullable"):
+        return replace(_from_native_dtype(native.type, target),
+                       native_dtype=NativeDtype(value=native.type, target=target))
+    if shape.item_shape is not None:
+        inner = getattr(native, "inner", getattr(native, "value_type", None))
+        if inner is not None:
+            shape = replace(shape, item_shape=_from_native_dtype(inner, target))
+    if shape.struct_fields:
+        fields = getattr(native, "fields", None)
+        if fields is None:
+            pairs = ((item.name, item.type) for item in native)
+        elif hasattr(fields, "items"):
+            pairs = fields.items()
+        else:
+            pairs = ((item.name, item.type if target is TypeTarget.PYARROW else item.dtype) for item in fields)
+        shape = replace(shape, struct_fields=tuple(
+            (name, _from_native_dtype(child, target)) for name, child in pairs
+        ))
+    return replace(shape, native_dtype=evidence)
+
+
 def _from_schema(schema: Any, target: TypeTarget) -> dict[str, SourceShape]:
     return {
-        name: _shape_from_dtype(_resolve(dtype, target))
+        name: _from_native_dtype(dtype, target)
         for name, dtype in schema.items()
     }
 
@@ -113,16 +142,21 @@ def _from_ibis_schema(schema: Any) -> dict[str, SourceShape]:
 
 def _from_narwhals_schema(native: Any) -> dict[str, SourceShape]:
     underlying = native.to_native()
-    if (
-        is_pandas_dataframe(underlying) or is_pyarrow_table(underlying)
-        or is_polars_dataframe(underlying) or is_polars_lazyframe(underlying)
-    ):
-        return extract_source_shapes(underlying)
-    return _from_schema(native.schema, TypeTarget.NARWHALS)
+    if is_pandas_dataframe(underlying):
+        from mountainash.core.lazy_imports import import_narwhals
+        schema = import_narwhals().Schema.from_pandas_like(underlying.dtypes.to_dict())
+        shapes = _from_pandas_dtypes(underlying.dtypes)
+        return {name: replace(shape, native_dtype=NativeDtype(
+            value=deepcopy(schema[name]), target=TypeTarget.NARWHALS,
+        )) for name, shape in shapes.items()}
+    return _from_schema(native.collect_schema(), TypeTarget.NARWHALS)
 
 
-def extract_source_shapes(native: Any) -> dict[str, SourceShape]:
+def extract_source_shapes(native: Any, *, target: TypeTarget | None = None) -> dict[str, SourceShape]:
     """Extract recursive source shapes using schema metadata only."""
+    if target is TypeTarget.NARWHALS and not (is_narwhals_dataframe(native) or is_narwhals_lazyframe(native)):
+        from mountainash.core.lazy_imports import import_narwhals
+        return _from_narwhals_schema(import_narwhals().from_native(native))
     if is_polars_dataframe(native) or is_polars_lazyframe(native):
         return _from_polars_schema(native.collect_schema())
     if is_pyarrow_table(native):
diff --git a/src/mountainash/typespec/spec.py b/src/mountainash/typespec/spec.py
index cc79c33d..5683a0e5 100644
--- a/src/mountainash/typespec/spec.py
+++ b/src/mountainash/typespec/spec.py
@@ -22,8 +22,8 @@
 from typing import Any, Dict, List, Optional, Union
 
 from mountainash.core.constants import DECIMAL_ROUNDING_MODES
-from mountainash.core.dtypes import DecimalDtype  # noqa: TC001 - runtime pydantic resolution via DataResource
-from mountainash.typespec.universal_types import UniversalType, parse_universal
+from mountainash.core.dtypes import DecimalDtype, MountainashDtype, NativeDtype, parse_cast_target  # noqa: TC001
+from mountainash.typespec.universal_types import UniversalType, from_canonical, parse_universal
 from mountainash.typespec.errors import (
     IncompatibleFieldPropertiesError,
     InvalidDescriptorRelationship,
@@ -149,14 +149,43 @@ class FieldSpec:
     object_fields: Optional[List["FieldSpec"]] = None  # x-mountainash: OBJECT inner-field schema
     item_object_fields: Optional[List["FieldSpec"]] = None  # x-mountainash: ARRAY item struct schema
     delimiter: Optional[str] = None  # valid only when type is LIST (lexical)
-    backend_type: Optional[str] = None
-    dtype: Optional[DecimalDtype] = None  # x-mountainash: portable exact decimal (item 241)
+    backend_type: Any = None  # str or native dtype; native objects normalize to NativeDtype
+    dtype: MountainashDtype | DecimalDtype | None = None
     null_fill: Any = None
     rename_from: Optional[str] = None
     custom_cast: Optional[str] = None
     rounding: str | None = None  # x-mountainash: conversion instruction, not dtype identity
 
     def __post_init__(self) -> None:
+        if self.dtype is not None:
+            if not isinstance(self.dtype, (MountainashDtype, DecimalDtype)):
+                raise InvalidDtypeDeclaration(self.name, "dtype requires a canonical enum or DecimalDtype", self.dtype)
+            broad, fmt = from_canonical(self.dtype)
+            if self.dtype in (MountainashDtype.JSON, MountainashDtype.DURATION):
+                broad = UniversalType.ANY
+            if self.type is UniversalType.ANY:
+                self.type = broad
+            elif self.type is UniversalType.GEOPOINT:
+                from mountainash.typespec.converters import _resolve_broad_field_canonical
+                if _resolve_broad_field_canonical(self) != self.dtype:
+                    raise IncompatibleFieldPropertiesError(self.name, "dtype", self.type, (broad,))
+            elif not (
+                self.type is broad
+                or (self.dtype is MountainashDtype.JSON and self.type is UniversalType.GEOJSON)
+                or (self.dtype is MountainashDtype.LIST and self.type is UniversalType.LIST)
+            ):
+                raise IncompatibleFieldPropertiesError(self.name, "dtype", self.type, (broad,))
+            if fmt is not None:
+                if self.format not in ("default", fmt):
+                    raise InvalidDtypeDeclaration(self.name, "dtype contradicts field format", self.format)
+                self.format = fmt
+            if self.dtype is MountainashDtype.STRING and self.format == "binary":
+                raise InvalidDtypeDeclaration(self.name, "string dtype contradicts binary format")
+        if self.backend_type is not None and not isinstance(self.backend_type, str):
+            native = parse_cast_target(self.backend_type)
+            if not isinstance(native, NativeDtype):
+                raise InvalidDtypeDeclaration(self.name, "backend_type requires a native dtype", self.backend_type)
+            self.backend_type = native
         if self.item_type is not None and self.type is not UniversalType.LIST:
             raise IncompatibleFieldPropertiesError(
                 self.name,
@@ -179,19 +208,10 @@ def __post_init__(self) -> None:
             raise IncompatibleFieldPropertiesError(
                 self.name, "object_fields", self.type, (UniversalType.OBJECT,)
             )
-        if self.dtype is not None:
-            if self.type is not UniversalType.NUMBER:
-                raise IncompatibleFieldPropertiesError(
-                    self.name, "dtype", self.type, (UniversalType.NUMBER,)
-                )
-            if self.backend_type is not None:
-                raise InvalidDtypeDeclaration(
-                    self.name, "dtype and backend_type are mutually exclusive"
-                )
-            if self.categories is not None:
-                raise InvalidDtypeDeclaration(
-                    self.name, "dtype and categories cannot be combined"
-                )
+        if isinstance(self.dtype, DecimalDtype) and self.categories is not None:
+            raise InvalidDtypeDeclaration(
+                self.name, "dtype and categories cannot be combined"
+            )
         if self.rounding is not None:
             if not isinstance(self.dtype, DecimalDtype):
                 raise InvalidDtypeDeclaration(self.name, "rounding requires a DecimalDtype")
@@ -402,8 +422,7 @@ def compare_specs(
         if source_field.type != target_field.type:
             diff.type_changes[name] = (source_field.type, target_field.type)
         elif source_field.dtype != target_field.dtype:
-            # Declared exact decimals differ in p/s (or only one side declares one).
-            # Fields without a dtype on either side are compared exactly as before.
+            # Compare authored identities, not backend-dependent effective targets.
             diff.type_changes[name] = (source_field.dtype or source_field.type,
                                        target_field.dtype or target_field.type)
 
diff --git a/src/mountainash/typespec/validation.py b/src/mountainash/typespec/validation.py
index 524362c3..022e25b8 100644
--- a/src/mountainash/typespec/validation.py
+++ b/src/mountainash/typespec/validation.py
@@ -21,6 +21,34 @@ class SchemaValidationError(MountainashError):
     pass
 
 
+def _matches_refinements(expected, shape) -> bool:
+    """Check precise nested targets without narrowing broad-only declarations."""
+    from mountainash.core.dtypes import DecimalDtype
+    from .converters import _resolve_field_target, _native_types_equal
+
+    if expected.backend_type:
+        if shape is None or shape.native_dtype is None:
+            return False
+        native = shape.native_dtype
+        effective = _resolve_field_target(expected, native.target)
+        return _native_types_equal(native.value, effective.value, native.target)
+    if expected.dtype is not None:
+        if shape is None:
+            return False
+        actual = shape.decimal if isinstance(expected.dtype, DecimalDtype) else shape.canonical_type
+        if actual != expected.dtype:
+            return False
+    children = dict(shape.struct_fields) if shape is not None else {}
+    if any(not _matches_refinements(child, children.get(child.name)) for child in expected.object_fields or ()):
+        return False
+    item = shape.item_shape if shape is not None else None
+    children = dict(item.struct_fields) if item is not None else {}
+    return all(
+        _matches_refinements(child, children.get(child.name))
+        for child in expected.item_object_fields or ()
+    )
+
+
 def validate_match(
     df: 'SupportedDataFrames',
     expected_schema: 'TypeSpec',
@@ -46,6 +74,10 @@ def validate_match(
     """
     from .extraction import extract_from_dataframe
     from .spec import compare_specs
+    from .source_shape import extract_source_shapes
+    from .converters import _resolve_field_target, _native_types_equal
+    from mountainash.core.dtypes import NativeDtype, MountainashDtype
+    from .universal_types import UniversalType
 
     errors = []
 
@@ -58,6 +90,32 @@ def validate_match(
 
     # Compare
     diff = compare_specs(actual_schema, expected_schema, check_constraints=False)
+    shapes = None
+    for expected in expected_schema.fields:
+        actual = actual_schema.get_field(expected.name)
+        if actual is None:
+            continue
+        if expected.backend_type:
+            if shapes is None:
+                shapes = extract_source_shapes(df)
+            native = shapes[expected.name].native_dtype
+            effective = _resolve_field_target(expected, native.target)
+            if isinstance(effective, NativeDtype) and _native_types_equal(
+                native.value, effective.value, native.target,
+            ):
+                diff.type_changes.pop(expected.name, None)
+            else:
+                diff.type_changes[expected.name] = (native.value, effective.value)
+        elif expected.dtype is None and (
+            expected.type is UniversalType.ANY
+            or (actual.type is expected.type and isinstance(actual.dtype, MountainashDtype))
+        ):
+            diff.type_changes.pop(expected.name, None)
+        if not expected.backend_type and (expected.object_fields or expected.item_object_fields):
+            if shapes is None:
+                shapes = extract_source_shapes(df)
+            if not _matches_refinements(expected, shapes.get(expected.name)):
+                diff.type_changes[expected.name] = (actual, expected)
 
     if diff.missing_columns:
         errors.append(f"Missing required columns: {diff.missing_columns}")
diff --git a/src/mountainash/validation/plan.py b/src/mountainash/validation/plan.py
index 3938e314..652877bc 100644
--- a/src/mountainash/validation/plan.py
+++ b/src/mountainash/validation/plan.py
@@ -209,6 +209,7 @@ def build_compiled_plan(spec: TypeSpec, checks: Sequence[Any]) -> CompiledValida
 
 def _thaw_dataclass_type(identifier: str) -> type[Any]:
     """Resolve only the closed TypeSpec declaration vocabulary."""
+    from mountainash.core.dtypes import DecimalDtype
     from mountainash.typespec.spec import (
         FieldConstraints,
         FieldSpec,
@@ -221,6 +222,7 @@ def _thaw_dataclass_type(identifier: str) -> type[Any]:
     allowed = {
         f"{cls.__module__}:{cls.__qualname__}": cls
         for cls in (
+            DecimalDtype,
             FieldConstraints,
             FieldSpec,
             ForeignKey,
@@ -238,21 +240,22 @@ def _thaw_dataclass_type(identifier: str) -> type[Any]:
 
 
 def _thaw_enum(module_name: str, class_name: str, value: Any) -> Any:
-    """Resolve only TypeSpec's closed universal-type enum."""
+    """Resolve only TypeSpec's closed declaration enum vocabulary."""
     from mountainash.typespec.universal_types import UniversalType
+    from mountainash.core.dtypes import MountainashDtype, TypeTarget
 
-    if (module_name, class_name) != (
-        UniversalType.__module__,
-        UniversalType.__qualname__,
-    ):
-        raise ValueError(
-            f"unsupported frozen enum declaration {module_name}:{class_name}"
-        )
-    return UniversalType(value)
+    for cls in (UniversalType, MountainashDtype, TypeTarget):
+        if (module_name, class_name) == (cls.__module__, cls.__qualname__):
+            return cls(value)
+    raise ValueError(f"unsupported frozen enum declaration {module_name}:{class_name}")
 
 
 def thaw_value(value: Any) -> Any:
     """Reconstruct one private mutable declaration copy from a frozen snapshot."""
+    from copy import deepcopy
+    from mountainash.core.dtypes import NativeDtype
+    if isinstance(value, NativeDtype):
+        return deepcopy(value)
     if isinstance(value, Mapping) and "__dataclass__" in value:
         cls = _thaw_dataclass_type(value["__dataclass__"])
         return cls(**{name: thaw_value(item) for name, item in value["fields"].items()})
diff --git a/tests/conform/cross_backend/test_relation_conform.py b/tests/conform/cross_backend/test_relation_conform.py
index f0e580a1..83f71f81 100644
--- a/tests/conform/cross_backend/test_relation_conform.py
+++ b/tests/conform/cross_backend/test_relation_conform.py
@@ -14,6 +14,78 @@
 
 
 
+@pytest.mark.parametrize("backend_name", ALL_BACKENDS)
+@pytest.mark.parametrize("dtype,values", [
+    (ma.MountainashDtype.I8, [-128, 127, None]),
+    (ma.MountainashDtype.U8, [0, 255, None]),
+    (ma.MountainashDtype.FP32, [1.5, -2.25, None]),
+])
+def test_canonical_conform_retains_declared_identity(backend_name, backend_factory, dtype, values, request):
+    if dtype in {ma.MountainashDtype.I8, ma.MountainashDtype.U8} and backend_name in {"pandas", "narwhals-pandas"}:
+        from pandas.errors import IntCastingNaNError
+        request.applymarker(pytest.mark.xfail(
+            strict=True, raises=IntCastingNaNError,
+            reason="252: native Narwhals NumPy-backed float/NaN -> integer cast loses nullable representation",
+        ))
+    source = backend_factory.create({"value": values}, backend_name)
+    rel = ma.relation(source).conform(TypeSpec(
+        fields=[FieldSpec(name="value", dtype=dtype)], missing_values=[],
+    ))
+    out = rel.to_polars()
+    assert out["value"].to_list() == values
+    # Native-first schema reports SQLite's physical affinities. The typed
+    # Polars/Arrow egress below must still retain the precise declaration.
+    native_dtype = (
+        ma.MountainashDtype.FP64 if dtype is ma.MountainashDtype.FP32 else ma.MountainashDtype.I64
+    ) if backend_name == "ibis-sqlite" else dtype
+    assert rel.schema["value"] is native_dtype
+    import polars as pl
+    assert out.schema["value"] == {
+        ma.MountainashDtype.I8: pl.Int8,
+        ma.MountainashDtype.U8: pl.UInt8,
+        ma.MountainashDtype.FP32: pl.Float32,
+    }[dtype]
+
+
+@pytest.mark.parametrize("backend_name", ALL_BACKENDS)
+def test_native_override_controls_conform_and_inference(backend_name, backend_factory, request):
+    if backend_name in {"pandas", "narwhals-pandas"}:
+        from pandas.errors import IntCastingNaNError
+        request.applymarker(pytest.mark.xfail(
+            strict=True, raises=IntCastingNaNError,
+            reason="252: native Narwhals NumPy-backed float/NaN -> integer cast loses nullable representation",
+        ))
+    source = backend_factory.create({"value": [127, None]}, backend_name)
+    override = "int64" if backend_name.startswith("ibis") else "Int64"
+    rel = ma.relation(source).conform(TypeSpec(
+        fields=[FieldSpec(name="value", dtype=ma.MountainashDtype.I8, backend_type=override)],
+        missing_values=[],
+    ))
+    with pytest.warns(UserWarning, match="value"):
+        out = rel.to_polars()
+    assert out["value"].to_list() == [127, None]
+    assert rel.schema["value"] is ma.MountainashDtype.I64
+    import polars as pl
+    assert out.schema["value"] == pl.Int64
+
+
+def test_explicit_native_timestamp_freeze_compares_parameters():
+    # Polars-owned parameters: other targets cannot consume this native object.
+    import polars as pl
+    from mountainash.conform.errors import SchemaDriftError
+    source = pl.DataFrame({"at": [1, 1001, None]}).with_columns(pl.col("at").cast(pl.Datetime("ns", "UTC")))
+    for unit in ("ns", "us"):
+        spec = TypeSpec(fields=[FieldSpec(
+            name="at", backend_type=pl.Datetime(unit, "UTC"),
+        )], missing_values=[])
+        rel = ma.relation(source).conform(spec, contract={"data_type": "freeze"})
+        if unit == "ns":
+            assert rel.to_polars()["at"].dt.epoch("ns").to_list() == [1, 1001, None]
+        else:
+            with pytest.raises(SchemaDriftError):
+                rel.to_polars()
+
+
 # ALL_BACKENDS = [
 #     "polars",
 #     "pandas",
@@ -684,3 +756,95 @@ def test_owns_contract_override(self, backend_name, backend_factory, as_proxy):
         # override while execution drops them (separate defect, not ownership).
         self._check(old, expected_cols=["value"], schema_parity=False)
         self._check(new, expected_cols=["value", "support"])
+
+
+def test_nested_override_freeze_uses_effective_child_type():
+    import polars as pl
+    spec = TypeSpec(fields=[FieldSpec(name="record", dtype=ma.MountainashDtype.STRUCT,
+        object_fields=[FieldSpec(name="id", dtype=ma.MountainashDtype.I8, backend_type=pl.Int64)],
+    )], missing_values=[])
+    source = pl.DataFrame({"record": [{"id": 300}, None]})
+    with pytest.warns(UserWarning):
+        out = ma.relation(source).conform(spec, contract={"data_type": "freeze"}).to_polars()
+    assert out.schema["record"] == pl.Struct({"id": pl.Int64})
+    assert out["record"].to_list() == [{"id": 300}, None]
+
+
+@pytest.mark.cross_backend
+@pytest.mark.parametrize("backend_name", ALL_BACKENDS)
+@pytest.mark.parametrize("action", ["coerce", "evolve", "freeze", "discard_value", "discard_row"])
+def test_native_override_reconciliation_policy(backend_name, backend_factory, action):
+    from mountainash.conform.errors import SchemaDriftError
+    from mountainash.core.types import BackendCapabilityError
+    from ibis.common.exceptions import OperationNotDefinedError
+    from fixtures.call_expectations import expect_call_failure
+    discarding = action in {"discard_value", "discard_row"}
+    values = ["1", "300", "bad"] if discarding else ["1", "300"]
+    source = backend_factory.create({"value": values}, backend_name)
+    override = "int64" if backend_name.startswith("ibis") else "Int64"
+    spec = TypeSpec(fields=[FieldSpec(
+        name="value", dtype=ma.MountainashDtype.I8, backend_type=override,
+    )], missing_values=[])
+    rel = ma.relation(source).conform(spec, contract={"data_type": action})
+    with pytest.warns(UserWarning, match="value"):
+        if action == "freeze":
+            with pytest.raises(SchemaDriftError):
+                rel.to_polars()
+            return
+        with expect_call_failure(
+            when=discarding and backend_name in {
+                "pandas", "narwhals-polars", "narwhals-pandas", "narwhals-lazy", "ibis-sqlite",
+            },
+            errors=(OperationNotDefinedError,) if backend_name == "ibis-sqlite" else (BackendCapabilityError,),
+            reason="Existing null-on-failure cast capability gap (known-divergences #19/#20)",
+        ):
+            out = rel.to_polars()
+    if action == "evolve":
+        assert out["value"].to_list() == ["1", "300"]
+        assert rel.schema == ma.relation(source).schema
+    else:
+        assert out["value"].to_list() == ([1, 300, None] if action == "discard_value" else [1, 300])
+        assert rel.schema["value"] is ma.MountainashDtype.I64
+
+
+def test_native_container_override_replaces_nested_casts():
+    # This native dtype is Polars-owned; other targets reject it.
+    import polars as pl
+    spec = TypeSpec(fields=[FieldSpec(name="record", dtype=ma.MountainashDtype.STRUCT,
+        object_fields=[FieldSpec(name="id", dtype=ma.MountainashDtype.I8)],
+        backend_type=pl.Struct({"id": pl.Int64}),
+    )], missing_values=[])
+    source = pl.DataFrame({"record": [{"id": 128}, None]})
+    with pytest.warns(UserWarning):
+        out = ma.relation(source).conform(spec).to_polars()
+    assert out.schema["record"] == pl.Struct({"id": pl.Int64})
+    assert out["record"].to_list() == [{"id": 128}, None]
+
+
+@pytest.mark.parametrize("values", [[], [None, None]])
+@pytest.mark.parametrize("in_array", [False, True])
+def test_null_typed_native_container_uses_declared_children(values, in_array):
+    # Polars Null is schema evidence; pandas object is deliberately not equivalent.
+    import polars as pl
+    child = FieldSpec(name="id", dtype=ma.MountainashDtype.I8)
+    field = FieldSpec(name="record", dtype=ma.MountainashDtype.LIST if in_array else ma.MountainashDtype.STRUCT,
+        **({"item_object_fields": [child]} if in_array else {"object_fields": [child]}))
+    out = ma.relation(pl.DataFrame({"record": values})).conform(
+        TypeSpec(fields=[field], missing_values=[]),
+    ).to_polars()
+    expected = pl.List(pl.Struct({"id": pl.Int8})) if in_array else pl.Struct({"id": pl.Int8})
+    assert out.schema["record"] == expected
+    assert out["record"].to_list() == values
+
+
+@pytest.mark.parametrize("action", ["discard_value", "discard_row"])
+def test_null_container_fill_obeys_failure_policy(action):
+    import polars as pl
+    field = FieldSpec(name="record", dtype=ma.MountainashDtype.STRUCT,
+        object_fields=[FieldSpec(name="id", dtype=ma.MountainashDtype.I8)],
+        null_fill={"id": 128})
+    out = ma.relation(pl.DataFrame({"record": [None]})).conform(
+        TypeSpec(fields=[field], missing_values=[]), contract={"data_type": action},
+    ).to_polars()
+    assert out.schema["record"] == pl.Struct({"id": pl.Int8})
+    assert out["record"].to_list() == [{"id": None}]
diff --git a/tests/conform/test_contract_data_type.py b/tests/conform/test_contract_data_type.py
index 975f414e..f9b4879c 100644
--- a/tests/conform/test_contract_data_type.py
+++ b/tests/conform/test_contract_data_type.py
@@ -47,7 +47,6 @@ def test_safe_cast_is_not_drift():
     assert out.drift.type_mismatches == []
     em = out.emitted[0]
     assert em.type_action == "coerce"
-    assert em.effective_type is None
 
 
 def test_unsafe_cast_coerce_reports_and_casts():
@@ -61,7 +60,6 @@ def test_unsafe_cast_coerce_reports_and_casts():
     assert out.drift.type_mismatches[0].applied is True
     em = out.emitted[0]
     assert em.type_action == "coerce"
-    assert em.effective_type is None
 
 
 def test_unsafe_cast_evolve_keeps_source_type():
diff --git a/tests/pydata/test_hybrid_conversion.py b/tests/pydata/test_hybrid_conversion.py
index ddf1e5e5..529b42c7 100644
--- a/tests/pydata/test_hybrid_conversion.py
+++ b/tests/pydata/test_hybrid_conversion.py
@@ -17,6 +17,15 @@
 from mountainash.typespec.universal_types import UniversalType
 
 
+
+def test_ingress_native_override_is_authoritative():
+    from mountainash import MountainashDtype
+    spec = TypeSpec(fields=[FieldSpec(name="id", dtype=MountainashDtype.I8, backend_type=pl.Int64)])
+    with pytest.warns(UserWarning):
+        out = apply_hybrid_conversion([{"id": 300}], spec)
+    assert out["id"].to_list() == [300]
+    assert out.schema["id"] == pl.Int64
+
 # ---------------------------------------------------------------------------
 # Fixtures
 # ---------------------------------------------------------------------------
diff --git a/tests/relations/test_schema_inference_conform.py b/tests/relations/test_schema_inference_conform.py
index 5f2e8066..56527194 100644
--- a/tests/relations/test_schema_inference_conform.py
+++ b/tests/relations/test_schema_inference_conform.py
@@ -620,3 +620,24 @@ def test_freeze_extra_columns_contract_does_not_raise_during_schema(self):
 
         with pytest.raises(SchemaDriftError):
             rel.to_polars()
+
+
+def test_mapping_tuple_conform_has_no_native_shape_requirement():
+    rel = ma.relation(({"x": 1},)).conform(TypeSpec.from_simple_dict({"x": "integer"}))
+    assert rel.schema["x"] is ma.MountainashDtype.I64
+    assert rel.to_polars()["x"].to_list() == [1]
+
+
+def test_narwhals_routed_native_subclass_uses_selected_dtype_domain():
+    import narwhals as nw
+
+    class Frame(pl.DataFrame):
+        pass
+
+    spec = TypeSpec(fields=[FieldSpec(name="id", dtype=ma.MountainashDtype.I8, backend_type=nw.Int64)])
+    rel = ma.relation(Frame({"id": [300]})).conform(spec, contract={"data_type": "freeze"})
+    assert rel.schema["id"] is ma.MountainashDtype.I64
+    with pytest.warns(UserWarning):
+        out = rel.to_polars()
+    assert out.schema["id"] == pl.Int64
+    assert out["id"].to_list() == [300]
diff --git a/tests/relations/test_schema_inference_types.py b/tests/relations/test_schema_inference_types.py
index fb9db0b1..bf3d3386 100644
--- a/tests/relations/test_schema_inference_types.py
+++ b/tests/relations/test_schema_inference_types.py
@@ -136,3 +136,22 @@ def test_resource_read_schema_unknown_for_malformed_or_absent_schema_evidence(fi
     assert infer_schema(ResourceReadRelNode(resource=resource)) == {
         "shape": SchemaTypeStatus.UNKNOWN
     }
+
+
+@pytest.mark.parametrize(
+    "extension,expected",
+    [
+        ({"dtype": {"kind": "canonical", "name": "i8"}}, D.I8),
+        ({"dtype": {"kind": "canonical", "name": "json"}}, D.JSON),
+        ({"dtype": {"kind": "canonical", "name": "duration"}}, D.DURATION),
+        ({"dtype": {"kind": "canonical", "name": "i8"}, "backend_type": "Int64"}, SchemaTypeStatus.UNKNOWN),
+    ],
+)
+def test_resource_descriptor_retains_precise_evidence_without_guessing_override_target(extension, expected):
+    from mountainash.relations.core.relation_nodes.extensions_mountainash import ResourceReadRelNode
+    from mountainash.typespec.datapackage import DataResource
+
+    resource = DataResource(name="data", data=[], table_schema={
+        "fields": [{"name": "value", "x-mountainash": extension}],
+    })
+    assert infer_schema(ResourceReadRelNode(resource=resource)) == {"value": expected}
diff --git a/tests/typespec/test_converters.py b/tests/typespec/test_converters.py
index 4a9a0438..10e95f6a 100644
--- a/tests/typespec/test_converters.py
+++ b/tests/typespec/test_converters.py
@@ -29,6 +29,81 @@
 from mountainash.typespec.spec import FieldSpec, TypeSpec
 
 
+@pytest.mark.parametrize("target", ["polars", "pandas", "arrow", "ibis"])
+def test_native_override_wins_over_precise_declaration(target):
+    import polars as pl
+    import pyarrow as pa
+    cases = {
+        "polars": (to_polars_schema, "Int64", pl.Int64),
+        "pandas": (to_pandas_dtypes, "Int64", "Int64"),
+        "arrow": (to_arrow_schema, "int64", pa.int64()),
+        "ibis": (to_ibis_schema, "int64", "int64"),
+    }
+    converter, override, expected = cases[target]
+    spec = TypeSpec(fields=[FieldSpec(name="x", dtype=D.I8, backend_type=override)])
+    with pytest.warns(UserWarning, match="x"):
+        schema = converter(spec)
+    actual = schema.field("x").type if target == "arrow" else schema["x"]
+    assert str(actual) == str(expected)
+
+
+def test_native_parameters_and_nested_canonical_children_survive_allocation():
+    import polars as pl
+    child = FieldSpec(name="at", dtype=D.TIMESTAMP, backend_type=pl.Datetime("ns", "UTC"))
+    spec = TypeSpec(fields=[FieldSpec(name="record", dtype=D.STRUCT, object_fields=[
+        FieldSpec(name="id", dtype=D.I8), child,
+    ])])
+    with pytest.warns(UserWarning):
+        frame = pl.DataFrame(schema=to_polars_schema(spec))
+    assert frame.schema["record"] == pl.Struct({"id": pl.Int8, "at": pl.Datetime("ns", "UTC")})
+
+
+def test_lossy_reverse_mapping_cannot_certify_native_equivalence():
+    import pyarrow as pa
+    spec = TypeSpec(fields=[FieldSpec(name="x", dtype=D.FP32, backend_type=pa.float16())])
+    with pytest.warns(UserWarning, match="x"):
+        assert to_arrow_schema(spec).field("x").type == pa.float16()
+
+
+def test_invalid_override_never_falls_back_to_canonical():
+    from mountainash.core.dtypes import InvalidBackendTypeError
+    spec = TypeSpec(fields=[FieldSpec(name="x", dtype=D.I8, backend_type="not-a-type")])
+    with pytest.raises(InvalidBackendTypeError):
+        to_arrow_schema(spec)
+
+
+def test_native_object_cannot_be_lowered_on_another_target():
+    import polars as pl
+    from mountainash.core.dtypes.errors import DtypeMappingError
+    spec = TypeSpec(fields=[FieldSpec(name="x", dtype=D.I8, backend_type=pl.Int8)])
+    with pytest.raises(DtypeMappingError):
+        to_arrow_schema(spec)
+
+
+def test_extracted_nested_width_is_portable_without_backend_spelling():
+    import polars as pl
+    import pyarrow as pa
+    from mountainash.typespec.extraction import extract_from_dataframe
+    frame = pl.DataFrame(schema={"record": pl.Struct({"id": pl.Int8, "payload": pl.Binary})})
+    spec = extract_from_dataframe(frame, preserve_backend_types=False)
+    restored = TypeSpec.from_frictionless(spec.to_frictionless())
+    assert to_arrow_schema(restored).field("record").type == pa.struct([
+        pa.field("id", pa.int8()), pa.field("payload", pa.binary()),
+    ])
+
+
+def test_schema_validation_uses_effective_native_target():
+    import polars as pl
+    from mountainash.typespec.validation import validate_match
+    spec = TypeSpec(fields=[FieldSpec(name="x", dtype=D.I8, backend_type="Int64")])
+    with pytest.warns(UserWarning):
+        assert validate_match(pl.DataFrame({"x": [127]}, schema={"x": pl.Int64}), spec) == (True, [])
+    with pytest.warns(UserWarning):
+        valid, errors = validate_match(pl.DataFrame({"x": [127]}, schema={"x": pl.Int8}), spec)
+    assert not valid
+    assert "x" in errors[0]
+
+
 # ============================================================================
 # Shared fixture schema
 # ============================================================================
@@ -282,14 +357,6 @@ def test_object_form_categories_use_shared_extraction(self):
         result = to_polars_schema(self._spec(cats, ordered=True))
         assert result["cat"] == pl.Enum([str(v) for v in categorical_values(cats)])
 
-    def test_categories_win_over_invalid_backend_type(self):
-        """Precedence (spec §5): a field with BOTH categories set AND an
-        invalid backend_type takes the categorical branch and never raises —
-        the backend_type is never even parsed for such a field."""
-        import polars as pl
-        result = to_polars_schema(self._spec(["a"], ordered=False, backend_type="garbage"))
-        assert result["cat"] is pl.Categorical
-
     def test_ibis_categories_stay_string(self):
         """Ibis has no categorical primitive — categories present still
         resolves to string (explicit, not silently untested)."""
@@ -598,6 +665,8 @@ def test_empty_or_none_backend_type_falls_through(self, backend_type):
             FieldSpec(name="x", type=UniversalType.INTEGER, backend_type=backend_type),
         ])
         assert to_polars_schema(spec)["x"] is pl.Int64
+        from mountainash.typespec.validation import validate_match
+        assert validate_match(pl.DataFrame({"x": [1]}), spec) == (True, [])
 
     def test_semantically_invalid_backend_type_raises_typed_error_not_raw(self):
         """Regression (GLM-5.2 whole-branch review, Important finding 1): a
@@ -724,3 +793,22 @@ def test_v2_six_field_spec_resolves_on_polars_narwhals_and_ibis(target) -> None:
     for name in ("geometry", "duration", "year", "yearmonth"):
         assert resolved[name] == string_type
         assert registry.from_native(resolved[name], target) is D.STRING
+
+
+@pytest.mark.parametrize("in_array", [False, True])
+def test_validate_match_checks_nested_effective_types(in_array):
+    import polars as pl
+    from mountainash.core.dtypes import MountainashDtype
+    from mountainash.typespec.validation import validate_match
+    child = FieldSpec(name="id", dtype=MountainashDtype.I8)
+    field = FieldSpec(name="record", dtype=MountainashDtype.LIST if in_array else MountainashDtype.STRUCT,
+        **({"item_object_fields": [child]} if in_array else {"object_fields": [child]}))
+    source = pl.DataFrame({"record": [[{"id": 128}]] if in_array else [{"id": 128}]})
+    spec = TypeSpec(fields=[field])
+    assert not validate_match(source, spec)[0]
+    child.backend_type = "Int64"
+    with pytest.warns(UserWarning):
+        assert validate_match(source, spec) == (True, [])
+    child.dtype = None
+    child.backend_type = None
+    assert validate_match(source, spec) == (True, [])
diff --git a/tests/typespec/test_decimal_field.py b/tests/typespec/test_decimal_field.py
index 789c2418..be06ede5 100644
--- a/tests/typespec/test_decimal_field.py
+++ b/tests/typespec/test_decimal_field.py
@@ -37,7 +37,6 @@ def test_default_is_none(self):
         [
             ({"type": UniversalType.STRING}, IncompatibleFieldPropertiesError),
             ({"type": UniversalType.INTEGER}, IncompatibleFieldPropertiesError),
-            ({"type": UniversalType.NUMBER, "backend_type": "Float64"}, InvalidDtypeDeclaration),
             ({"type": UniversalType.NUMBER, "categories": [1.5]}, InvalidDtypeDeclaration),
         ],
     )
diff --git a/tests/typespec/test_declaration_fingerprint.py b/tests/typespec/test_declaration_fingerprint.py
index 6ecb4afa..99d35d14 100644
--- a/tests/typespec/test_declaration_fingerprint.py
+++ b/tests/typespec/test_declaration_fingerprint.py
@@ -1,6 +1,7 @@
 """Declaration fingerprint compatibility tests."""
 from __future__ import annotations
 
+import pytest
 from mountainash.typespec._fingerprint import declaration_fingerprint
 from mountainash.typespec.spec import FieldSpec, TypeSpec
 from mountainash.typespec.universal_types import UniversalType
@@ -24,3 +25,30 @@ def test_equivalent_typespec_declarations_have_matching_fingerprints():
     assert declaration_fingerprint(freeze_typespec(first)) == declaration_fingerprint(
         freeze_typespec(second)
     )
+
+
+@pytest.mark.parametrize("backend", ["polars", "narwhals", "arrow", "pandas", "ibis"])
+def test_native_parameter_identity_is_stable_and_distinguishes_changes(backend):
+    import polars as pl
+    import narwhals as nw
+    import pandas as pd
+    import pyarrow as pa
+    import ibis.expr.datatypes as dt
+
+    constructors = {
+        "polars": lambda changed: pl.Struct({"at": pl.Datetime("us" if changed else "ns", "UTC")}),
+        "narwhals": lambda changed: nw.Datetime("us" if changed else "ns", "UTC"),
+        "arrow": lambda changed: pa.list_(pa.int16() if changed else pa.int8()),
+        "pandas": lambda changed: pd.CategoricalDtype(
+            [f"category-{i}" for i in range(100)] + ["changed" if changed else "original"], ordered=True,
+        ),
+        "ibis": lambda changed: dt.Array(dt.int16 if changed else dt.int8),
+    }
+
+    def fingerprint(changed):
+        spec = TypeSpec(fields=[FieldSpec(name="value", backend_type=constructors[backend](changed))])
+        return build_compiled_plan(spec, ()).declaration_fingerprint
+
+    first = fingerprint(False)
+    assert fingerprint(False) == first
+    assert fingerprint(True) != first
diff --git a/tests/typespec/test_frictionless.py b/tests/typespec/test_frictionless.py
index 417c513c..89191a06 100644
--- a/tests/typespec/test_frictionless.py
+++ b/tests/typespec/test_frictionless.py
@@ -8,6 +8,13 @@
 
 import pytest
 
+import polars as pl
+import pyarrow as pa
+
+from mountainash.core.dtypes import MountainashDtype
+from mountainash.typespec.converters import to_arrow_schema
+from mountainash.typespec.errors import InvalidDtypeDeclaration, IncompatibleFieldPropertiesError
+
 from mountainash.typespec.errors import (
     InvalidFieldMatchDeclaration,
     InvalidKeyShapeError,
@@ -30,6 +37,75 @@
     from pathlib import Path
 
 
+
+@pytest.mark.parametrize("dtype", list(MountainashDtype))
+def test_canonical_descriptor_identity_survives_json(dtype):
+    descriptor = {"fields": [{"name": "value", "x-mountainash": {
+        "dtype": {"kind": "canonical", "name": dtype.value},
+    }}]}
+    spec = typespec_from_frictionless(descriptor)
+    restored = typespec_from_frictionless(json.loads(json.dumps(spec.to_frictionless())))
+    assert restored.fields[0].dtype is dtype
+    if dtype in (MountainashDtype.JSON, MountainashDtype.DURATION):
+        assert "type" not in spec.to_frictionless()["fields"][0]
+
+
+@pytest.mark.parametrize("name,broad,native", [
+    ("i8", "integer", pa.int8()), ("i16", "integer", pa.int16()),
+    ("i32", "integer", pa.int32()), ("i64", "integer", pa.int64()),
+    ("u8", "integer", pa.uint8()), ("u16", "integer", pa.uint16()),
+    ("u32", "integer", pa.uint32()), ("u64", "integer", pa.uint64()),
+    ("fp32", "number", pa.float32()), ("fp64", "number", pa.float64()),
+    ("binary", "string", pa.binary()),
+])
+@pytest.mark.parametrize("explicit", [False, True])
+def test_precise_declarations_materialize_authored_width(name, broad, native, explicit):
+    kwargs = {"type": UniversalType(broad)} if explicit else {}
+    field = FieldSpec(name="value", dtype=MountainashDtype(name), **kwargs)
+    descriptor = TypeSpec(fields=[field]).to_frictionless()
+    assert descriptor["fields"][0]["type"] == broad
+    spec = typespec_from_frictionless(json.loads(json.dumps(descriptor)))
+    assert to_arrow_schema(spec).field("value").type == native
+
+
+@pytest.mark.parametrize("container", [None, "object_fields", "item_object_fields"])
+def test_native_objects_cannot_escape_through_nested_descriptors(container):
+    child = FieldSpec(name="at", backend_type=pl.Datetime("ns", "UTC"))
+    if container:
+        broad = UniversalType.OBJECT if container == "object_fields" else UniversalType.ARRAY
+        field = FieldSpec(name="outer", type=broad, **{container: [child]})
+    else:
+        field = child
+    with pytest.raises(InvalidDtypeDeclaration, match="at"):
+        TypeSpec(fields=[field]).to_frictionless()
+    assert child.backend_type.value == pl.Datetime("ns", "UTC")
+
+
+@pytest.mark.parametrize("dtype,broad", [
+    (MountainashDtype.I8, UniversalType.STRING),
+    (MountainashDtype.FP32, UniversalType.INTEGER),
+    (MountainashDtype.DURATION, UniversalType.DURATION),
+    (MountainashDtype.STRING, UniversalType.GEOJSON),
+])
+def test_contradictory_canonical_descriptor_is_rejected(dtype, broad):
+    with pytest.raises(IncompatibleFieldPropertiesError):
+        typespec_from_frictionless({"fields": [{
+            "name": "value", "type": broad.value,
+            "x-mountainash": {"dtype": {"kind": "canonical", "name": dtype.value}},
+        }]})
+
+
+@pytest.mark.parametrize("raw", [
+    {"kind": "canonical", "name": "int8"},
+    {"kind": "canonical", "name": "unknown"},
+    {"kind": "canonical"}, {"kind": "canonical", "name": []},
+])
+def test_malformed_canonical_descriptor_is_rejected(raw):
+    with pytest.raises(InvalidDtypeDeclaration):
+        typespec_from_frictionless({"fields": [{"name": "value", "x-mountainash": {
+            "dtype": raw,
+        }}]})
+
 # ============================================================================
 # TestToFrictionless
 # ============================================================================
diff --git a/tests/validation/test_plan.py b/tests/validation/test_plan.py
index da1a8a65..3c2e0d66 100644
--- a/tests/validation/test_plan.py
+++ b/tests/validation/test_plan.py
@@ -64,6 +64,39 @@ def _run(plan, data):
     )
 
 
+@pytest.mark.parametrize("dtype,value", [
+    (ma.MountainashDtype.I8, 127),
+    (ma.DecimalDtype(5, 2), 12),
+])
+def test_precise_declarations_execute_through_compiled_validation(dtype, value):
+    import polars as pl
+    field = FieldSpec(name="value", dtype=dtype,
+                      rounding="half_to_even" if isinstance(dtype, ma.DecimalDtype) else None)
+    plan = build_compiled_plan(TypeSpec(fields=[field], missing_values=[]), [
+        RowRule(id="value_kept", expr=ma.col("value").eq(value)),
+    ])
+    field.dtype = ma.MountainashDtype.BOOL
+    for _ in range(2):
+        assert _run(plan, pl.DataFrame({"value": [value]})).passes
+
+
+def test_native_validation_snapshot_owns_executable_parameters():
+    # Native Polars override ownership is exercised on its owning backend.
+    import datetime
+    import polars as pl
+    native = pl.Datetime("ns")
+    field = FieldSpec(name="at", backend_type=native)
+    spec = TypeSpec(fields=[field], missing_values=[])
+    checks = [RowRule(id="instant", expr=ma.col("at").eq(datetime.datetime(1970, 1, 1, microsecond=1000)))]
+    old = build_compiled_plan(spec, checks)
+    native.time_unit = "us"
+    new = build_compiled_plan(spec, checks)
+    data = pl.DataFrame({"at": [1_000_000]})
+    assert not _run(new, data).passes
+    for _ in range(2):
+        assert _run(old, data).passes
+
+
 def _threshold():
     expr = ma.col("v").lt(ma.lit(10))