fix(hashing): route Schema through SchemaHandler, not _expand_mapping (ITL-639)

src/orcapod/contexts/data/v0.1.json

@@ -120,7 +120,8 @@
                 [{"_type": "pyarrow.RecordBatch"},      {"_class": "orcapod.hashing.semantic_hashing.builtin_handlers.ArrowTableHandler",   "_config": {}}],
                 [{"_type": "numpy.ndarray"},            {"_class": "orcapod.hashing.semantic_hashing.builtin_handlers.NumpyArrayHandler",        "_config": {}}],
                 [{"_type": "pandas.DataFrame"},         {"_class": "orcapod.hashing.semantic_hashing.builtin_handlers.PandasDataFrameHandler",  "_config": {}}],
-                [{"_type": "pandas.Series"},            {"_class": "orcapod.hashing.semantic_hashing.builtin_handlers.PandasSeriesHandler",      "_config": {}}]
+                [{"_type": "pandas.Series"},            {"_class": "orcapod.hashing.semantic_hashing.builtin_handlers.PandasSeriesHandler",      "_config": {}}],
+                [{"_type": "orcapod.types.Schema"},     {"_class": "orcapod.hashing.semantic_hashing.builtin_handlers.SchemaHandler",            "_config": {"type_converter": {"_ref": "type_converter"}}}]
             ]
         }
     },
@@ -160,7 +161,8 @@
             "Added spikeinterface.core.motion.Motion as a native value type via LogicalSIMotion (large_binary/.npz) and SIMotionHandler (SHA-256 of .npz bytes); auto-registered when spikeinterface is installed via _optional entries in v0.1.json (ITL-470)",
             "Added spikeinterface.core.SortingAnalyzer as a native value type via LogicalSISortingAnalyzer (large_string/JSON path reference {folder, format}) and SISortingAnalyzerHandler (SHA-256 of folder path string, phase 1); auto-registered when spikeinterface is installed via _optional entries in v0.1.json (ITL-469)",
             "Wired file_hasher reference into BasicDirectoryHasher so op.Directory hashing uses the same FileContentHasherProtocol as op.File hashing; required file_hasher constructor arg enforces explicit injection",
-            "Moved SpikeInterface types (LogicalSIRecording, LogicalSISorting, LogicalSIMotion, LogicalSISortingAnalyzer) and their handlers to the standalone orcapod-extension-spikeinterface package; removed all SI _optional entries from this spec and SI deps from pyproject.toml; introduced OrcapodExtension protocol and op.register_extension() for normalized extension registration (ITL-473)"
+            "Moved SpikeInterface types (LogicalSIRecording, LogicalSISorting, LogicalSIMotion, LogicalSISortingAnalyzer) and their handlers to the standalone orcapod-extension-spikeinterface package; removed all SI _optional entries from this spec and SI deps from pyproject.toml; introduced OrcapodExtension protocol and op.register_extension() for normalized extension registration (ITL-473)",
+            "Registered SchemaHandler for orcapod.types.Schema in the context registry; moved handler dispatch before _is_structure check in hash_object so SchemaHandler takes priority over _expand_mapping for Schema objects (ITL-639)"
         ]
     }
 }

src/orcapod/hashing/semantic_hashing/builtin_handlers.py

@@ -184,14 +184,73 @@ def handle(self, obj: Any, hasher: "SemanticHasherProtocol") -> ContentHash:
 
 
 class SchemaHandler:
-    """Hasher for ``Schema`` objects."""
+    """Hasher for ``Schema`` objects.
+
+    Canonical, explicit path for schema hashing. For each field, hashes the Python
+    type via ``hasher.hash_object`` (which dispatches to ``TypeObjectHandler``, using
+    the stable ``logical_type_name`` / Arrow extension name for registered types and the
+    stable ``builtins.*`` / stdlib path for native types). Field names are sorted so the
+    hash is deterministic regardless of insertion order.
+
+    This produces the same hash as the previous accidental path through ``_expand_mapping``
+    (Schema is a Mapping), preserving all existing hash values.
+
+    ``Schema.optional_fields`` is intentionally excluded from the hash. Two schemas with
+    the same field names and types but different optionality are hash-equivalent.
+    Optionality is a Python-level execution contract (which parameters have defaults),
+    not part of the structural identity used for caching or pipeline routing.
+
+    When ``type_converter`` is provided, every field type is verified to be
+    Arrow-translatable before hashing. A type that cannot be converted raises
+    ``TypeError`` with a diagnostic message, catching unregistered types at hash time
+    rather than silently falling back to a potentially unstable Python module path.
+
+    Args:
+        type_converter: Optional ``TypeConverterProtocol``. When provided, validates
+            Arrow-translatability per field. When ``None`` (e.g. in tests without a
+            full ``DataContext``), validation is skipped.
+    """
+
+    def __init__(self, type_converter: "TypeConverterProtocol | None" = None) -> None:
+        self._type_converter = type_converter
 
     def handle(self, obj: Any, hasher: "SemanticHasherProtocol") -> Any:
+        """Hash ``obj`` by hashing each field type individually and sorting by name.
+
+        Args:
+            obj: A ``Schema`` instance.
+            hasher: The calling ``SemanticHasherProtocol`` — used to hash each field's
+                Python type.
+
+        Returns:
+            A ``dict[str, str]`` mapping field name to the ``to_string()`` of each
+            field type's hash, sorted by field name for determinism.
+
+        Raises:
+            TypeError: If ``obj`` is not a ``Schema``, or if ``type_converter`` is
+                provided and a field type is not Arrow-translatable.
+        """
         if not isinstance(obj, Schema):
             raise TypeError(
                 f"SchemaHandler: expected a Schema, got {type(obj)!r}"
             )
-        raise NotImplementedError("SchemaHandler is not yet implemented.")
+        result: dict[str, str] = {}
+        for field_name, python_type in obj.items():
+            if self._type_converter is not None:
+                try:
+                    self._type_converter.python_type_to_arrow_type(python_type)
+                except (TypeError, ValueError) as exc:
+                    raise TypeError(
+                        f"SchemaHandler: field {field_name!r} has type "
+                        f"{python_type!r} that is not Arrow-translatable. "
+                        f"Every type in a schema must be Arrow-convertible — "
+                        f"register it as an orcapod logical type or use a "
+                        f"supported native type (int, str, float, bool, bytes, "
+                        f"datetime, date)."
+                    ) from exc
+            result[field_name] = hasher.hash_object(python_type).to_string()
+        # Sort by field name for determinism — matches _expand_mapping's sort_keys.
+        return dict(sorted(result.items()))
 
 
 class FileHandler:
@@ -469,10 +528,11 @@ def register_builtin_python_type_handlers(
         directory_hasher: Optional ``DirectoryHasherProtocol`` for directory tree hashing.
             Defaults to ``BasicDirectoryHasher(sha256)``.
         type_converter: Optional ``TypeConverterProtocol`` forwarded to
-            ``TypeObjectHandler`` and ``FunctionSignatureExtractor`` for stable
-            canonical type-name resolution via ``get_logical_type()``.
-            When ``None`` (the default), both handlers fall back to the raw
-            ``"type:<module>.<qualname>"`` serialisation.
+            ``TypeObjectHandler``, ``FunctionSignatureExtractor``, and
+            ``SchemaHandler`` for stable canonical type-name resolution via
+            ``get_logical_type()`` and Arrow-translatability validation.
+            When ``None`` (the default), all three handlers fall back to the raw
+            ``"type:<module>.<qualname>"`` serialisation and skip Arrow validation.
     """
     if file_hasher is None:
         from orcapod.hashing.file_hashers import FileHasher
@@ -526,7 +586,7 @@ def register_builtin_python_type_handlers(
     except AttributeError:
         pass
 
-    registry.register(Schema, SchemaHandler())
+    registry.register(Schema, SchemaHandler(type_converter=type_converter))
 
     import pyarrow as _pa
     arrow_table_hasher = ArrowTableHandler(arrow_hasher)

src/orcapod/hashing/semantic_hashing/semantic_hasher.py

@@ -169,16 +169,8 @@ def hash_object(
         if isinstance(obj, (type(None), bool, int, float, str)):
             return self._hash_to_content_hash(obj)
 
-        # Structures: expand into a tagged tree, then hash the tree.
-        if _is_structure(obj):
-            expanded = self._expand_structure(
-                obj, _visited=frozenset(), resolver=resolver
-            )
-            return self._hash_to_content_hash(expanded)
-
-        # Semantic hasher dispatch: handler returns a representative Python structure
-        # (or a ContentHash as terminal); feed the result back into hash_object so
-        # that returning a plain structure is equivalent to calling hash_object on it.
+        # Registered handler dispatch (BEFORE _is_structure — ensures SchemaHandler
+        # takes priority over _expand_mapping for Schema objects, which are Mappings).
         handler = self._registry.get_handler(obj)
         if handler is not None:
             logger.debug(
@@ -189,6 +181,13 @@ def hash_object(
             result = handler.handle(obj, self)
             return self.hash_object(result, resolver=resolver)
 
+        # Structures: expand into a tagged tree, then hash the tree.
+        if _is_structure(obj):
+            expanded = self._expand_structure(
+                obj, _visited=frozenset(), resolver=resolver
+            )
+            return self._hash_to_content_hash(expanded)
+
         # ContentIdentifiableProtocol: use resolver if provided, else content_hash().
         if isinstance(obj, hp.ContentIdentifiableProtocol):
             if resolver is not None: