fix(data): support symbolic supervised projections

Build inert Map/Project/Select definitions when as_supervised receives a Definition or reference authority, while retaining live Map construction for materialized Dataset inputs.\n\nAccept Mat/Ref assertions without materializing their targets, document the behavior, and cover definition plus persisted-reference construction.\n\nVerified with the focused transform suite, affected model integration tests, and fatal flake8 checks.

Author: krafczyk

docs/data.md

@@ -325,11 +325,16 @@ Utility functions help with common supervised-learning structures:
 
 `as_supervised(dataset, inputs, targets)` builds a persisted ordinary Dataset
 projection yielding `(inputs, targets)`, rather than asking trainers to select
-fields. A scalar path selects one branch (and can retain a nested dictionary);
-named/nested selections use dictionaries of `Select.from_path(...)` leaves.
-Tuple/list selection trees require explicit `Select.from_path` leaves so path and
-output-tree intent cannot be guessed. `input_as_target=True` produces an
-autoencoder-style pair without copies or implicit batching.
+fields. A live Dataset returns a live `Map`. A soft Definition,
+ConcreteDefinition, ObjectRef, StateRef, or explicit `Ref(...)`/`Mat(...)`
+assertion returns an inert `Map` Definition without resolving or materializing
+the source; an assertion contributes its retained target to that new materializing
+Dataset graph. A scalar path selects one branch (and can retain a nested
+dictionary); named/nested selections use dictionaries of
+`Select.from_path(...)` leaves. Tuple/list selection trees require explicit
+`Select.from_path` leaves so path and output-tree intent cannot be guessed.
+`input_as_target=True` produces an autoencoder-style pair without copies or
+implicit batching.
 
 Native trainers consume this authored Dataset as-is. Put `Batch`, `Shuffle`,
 `Take`, and related controls in the Dataset graph before constructing an

src/dryml/data/util.py

@@ -17,10 +17,12 @@ def _xy_dataset(dataset, *, x_path=0, y_path=1):
 
 
 def as_supervised(dataset, inputs, targets=None, *, input_as_target: bool = False):
-    """Return a persisted Dataset graph yielding canonical ``(inputs, targets)`` pairs.
+    """Return a Dataset graph yielding canonical ``(inputs, targets)`` pairs.
 
     Args:
-        dataset: Source Dataset with a declared element specification.
+        dataset: Source Dataset with a declared element specification, or a soft
+            Definition, ConcreteDefinition, ObjectRef, StateRef, or explicit
+            ``Ref(...)``/``Mat(...)`` assertion expected to identify one.
         inputs: One scalar path, :class:`Select`, or nonempty tree of explicit
             ``Select`` leaves describing model inputs.
         targets: Matching target selection. It is required unless
@@ -29,56 +31,83 @@ def as_supervised(dataset, inputs, targets=None, *, input_as_target: bool = Fals
             autoencoder-style Dataset without copying or backend conversion.
 
     Returns:
-        An ordinary :class:`~dryml.data.dataset.Map` Dataset whose values are
-        ``(inputs, targets)`` and whose Method tree is persisted with the graph.
+        An ordinary :class:`~dryml.data.dataset.Map` Dataset for a live source,
+        or an inert Map Definition for a symbolic source. Its values are
+        ``(inputs, targets)`` and its Method tree is persisted with the graph.
 
     Raises:
-        TypeError: If ``dataset`` is not a Dataset or a selection tree is malformed.
+        TypeError: If ``dataset`` is not a Dataset or supported Dataset authority,
+            or a selection tree is malformed. A symbolic source that does not
+            materialize a Dataset fails at its normal Definition
+            binding/materialization boundary.
         ValueError: If a selection tree is empty or targets conflict with
             ``input_as_target``.
 
     Side Effects:
-        None. This helper does not open the Dataset, copy values, select a backend,
-        or add batching.
+        None. This helper does not open or materialize the Dataset, copy values,
+        select a backend, or add batching.
     """
 
+    from dryml.core import ConcreteDefinition, DefLink, Definition, ObjectRef, StateRef
     from dryml.data.dataset import Dataset, Map
 
-    if not isinstance(dataset, Dataset):
-        raise TypeError("as_supervised requires a Dataset.")
+    symbolic_types = (Definition, ConcreteDefinition, ObjectRef, StateRef)
+    symbolic = isinstance(dataset, (*symbolic_types, DefLink))
+    if not isinstance(dataset, Dataset) and not symbolic:
+        raise TypeError("as_supervised requires a Dataset or Dataset reference.")
+    if isinstance(dataset, DefLink):
+        dataset = dataset.target
+        if not isinstance(dataset, (Dataset, *symbolic_types)):
+            raise TypeError("as_supervised received an unsupported Ref/Mat target.")
     if type(input_as_target) is not bool:
         raise TypeError("input_as_target must be an exact bool.")
-    input_method = _selection_method(inputs)
+    input_method = _selection_method(inputs, symbolic=symbolic)
     if input_as_target:
         if targets is not None:
             raise ValueError("input_as_target does not accept an explicit target selection.")
         target_method = input_method
     else:
         if targets is None:
             raise ValueError("as_supervised requires targets unless input_as_target is true.")
-        target_method = _selection_method(targets)
-    return Map(dataset, Project(input_method, target_method), preserves_examples=True)
-
-
-def _selection_method(selection, *, tree_leaf: bool = False):
+        target_method = _selection_method(targets, symbolic=symbolic)
+    project = (
+        Project.defn(input_method, target_method)
+        if symbolic else Project(input_method, target_method)
+    )
+    if symbolic:
+        return Map.defn(dataset, project, preserves_examples=True)
+    return Map(dataset, project, preserves_examples=True)
+
+
+def _selection_method(
+        selection, *, tree_leaf: bool = False, symbolic: bool = False,
+):
     """Build one Project-compatible Method tree while rejecting tuple/list ambiguity."""
 
     if isinstance(selection, Select):
-        return selection
+        return Select.defn(selection.idxs) if symbolic else selection
     if isinstance(selection, (str, int)):
         if tree_leaf:
             raise TypeError("Selection trees require Select.from_path leaves.")
-        return Select.from_path((selection,))
+        return Select.defn(selection) if symbolic else Select.from_path((selection,))
     if isinstance(selection, dict):
         if not selection:
             raise ValueError("Selection trees must not be empty.")
-        return Project({key: _selection_method(branch, tree_leaf=True) for key, branch in selection.items()})
+        branches = {
+            key: _selection_method(branch, tree_leaf=True, symbolic=symbolic)
+            for key, branch in selection.items()
+        }
+        return Project.defn(branches) if symbolic else Project(branches)
     if isinstance(selection, (tuple, list)):
         if not selection:
             raise ValueError("Selection trees must not be empty.")
         if any(not isinstance(branch, Select) for branch in selection):
             raise TypeError("Ambiguous tuple/list selections require Select.from_path leaves.")
-        return Project(type(selection)(_selection_method(branch, tree_leaf=True) for branch in selection))
+        branches = type(selection)(
+            _selection_method(branch, tree_leaf=True, symbolic=symbolic)
+            for branch in selection
+        )
+        return Project.defn(branches) if symbolic else Project(branches)
     raise TypeError("Selections must be a scalar path, Select, or tree of Select leaves.")
 
 

tests/data/test_transform_nodes.py

@@ -4,6 +4,7 @@
 import numpy as np
 import pytest
 
+from dryml.core import Definition, Mat, Ref
 from dryml.core.cardinality import Cardinality
 from dryml.core.backend import Backend
 from dryml.core.tensor_spec import Dynamic, TensorSpec
@@ -773,6 +774,31 @@ def test_as_supervised_projects_scalar_whole_tree_named_and_autoencoder_targets(
     assert autoencoder_value[0] is autoencoder_value[1]
 
 
+@pytest.mark.parametrize("authority", ("soft", "concrete", "mat", "ref"))
+def test_as_supervised_authors_definition_native_autoencoder_projection(authority):
+    """Symbolic Dataset sources remain inert until their Map graph is built."""
+
+    source = Definition(
+        ListDataset,
+        [{"cart": np.array([1, 2])}],
+        {"cart": TensorSpec("int64", shape=(2,), backend="numpy")},
+    )
+    if authority == "concrete":
+        source = source.concretize()
+    elif authority == "mat":
+        source = Mat(source)
+    elif authority == "ref":
+        source = Ref(source)
+
+    projected = as_supervised(source, "cart", input_as_target=True)
+
+    assert isinstance(projected, Definition)
+    assert projected.cls is Map
+    inputs, targets = next(iter(projected.build()))
+    assert inputs is targets
+    assert inputs.tolist() == [1, 2]
+
+
 def test_as_supervised_rejects_empty_or_ambiguous_selection_trees():
     source = ListDataset(
         [{"x": np.array([1]), "y": np.array([2])}],
@@ -792,6 +818,35 @@ def test_as_supervised_rejects_empty_or_ambiguous_selection_trees():
         as_supervised(source, "x")
     with pytest.raises(ValueError, match="does not accept"):
         as_supervised(source, "x", "y", input_as_target=True)
+    with pytest.raises(TypeError, match="Dataset or Dataset reference"):
+        as_supervised(object(), "x", "y")
+
+
+def test_as_supervised_authors_graph_from_persisted_dataset_references(tmp_path):
+    """ObjectRef and StateRef inputs stay inert until the authored graph is built."""
+
+    from dryml.core import Repo
+    from dryml.core.store.dir import DirStore
+
+    repo = Repo(DirStore(tmp_path / "store"))
+    source = ListDataset(
+        [{"x": np.array([1]), "y": np.array([2])}],
+        {
+            "x": TensorSpec("int64", shape=(1,), backend="numpy"),
+            "y": TensorSpec("int64", shape=(1,), backend="numpy"),
+        },
+    )
+    state_ref = repo.save_object(source, deep_capture=True)
+
+    for reference in (state_ref.object, state_ref):
+        projected = as_supervised(reference, "x", "y")
+
+        assert isinstance(projected, Definition)
+        assert projected.cls is Map
+
+    inputs, targets = next(iter(as_supervised(state_ref, "x", "y").build(repo=repo)))
+    assert inputs.tolist() == [1]
+    assert targets.tolist() == [2]
 
 
 def test_as_supervised_nested_and_tuple_selection_trees_survive_save_load(tmp_path):