Feat session preprocessing

DashAI/alembic/versions/f4a91c62d8e7_add_preprocessing_to_model_session.py

@@ -0,0 +1,60 @@
+"""add preprocessing columns to model_session
+
+Revision ID: f4a91c62d8e7
+Revises: m6n7o8p9q0r1
+Create Date: 2026-09-08 10:00:00.000000
+
+Sessions created before this column existed have no preprocessing, so an
+empty/None value means exactly that — no backfill is needed.
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "f4a91c62d8e7"
+down_revision: Union[str, None] = "m6n7o8p9q0r1"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+_NEW_COLUMNS = [
+    ("preprocessing", sa.JSON(), True, None),
+    ("input_column_refs", sa.JSON(), True, None),
+    ("preprocessing_status", sa.String(), False, "ready"),
+    ("preprocessing_error", sa.String(), True, None),
+    ("preprocessing_artifacts_path", sa.String(), True, None),
+    ("preprocessing_job_id", sa.String(), True, None),
+]
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    if "model_session" not in inspector.get_table_names():
+        return
+    existing = {column["name"] for column in inspector.get_columns("model_session")}
+    for name, col_type, nullable, server_default in _NEW_COLUMNS:
+        if name in existing:
+            continue
+        op.add_column(
+            "model_session",
+            sa.Column(
+                name,
+                col_type,
+                nullable=nullable,
+                server_default=server_default,
+            ),
+        )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    if "model_session" not in inspector.get_table_names():
+        return
+    existing = {column["name"] for column in inspector.get_columns("model_session")}
+    for name, _, _, _ in _NEW_COLUMNS:
+        if name not in existing:
+            continue
+        op.drop_column("model_session", name)

DashAI/back/api/api_v1/endpoints/model_sessions.py

@@ -14,6 +14,8 @@
 )
 from DashAI.back.api.utils import remove_path
 from DashAI.back.dependencies.database.models import Dataset, ModelSession, Run
+from DashAI.back.job.preprocessing_job import PreprocessingJob
+from DashAI.back.preprocessing.column_ref import ConverterSequence, RawColumnRef
 from DashAI.back.splitters.splits_payload import (
     META_KEYS,
     normalize_splits_payload,
@@ -23,6 +25,7 @@
 if TYPE_CHECKING:
     from sqlalchemy.orm import sessionmaker
 
+    from DashAI.back.dependencies.job_queues import BaseJobQueue
     from DashAI.back.dependencies.registry import ComponentRegistry
     from DashAI.back.tasks.base_task import BaseTask
 
@@ -32,6 +35,22 @@
 
 router = APIRouter()
 
+# Every concrete value type name that DashAIValue's wildcard covers (i.e. every
+# DashAIValue subclass in DashAI/back/types/value_types.py), used so a task
+# declaring the wildcard (e.g. ClassificationTask) still accepts a converter
+# group whose declared type is one of these, not just "DashAIValue" itself.
+_DASHAI_VALUE_TYPE_NAMES = {
+    "Integer",
+    "Float",
+    "Text",
+    "Time",
+    "Timestamp",
+    "Duration",
+    "Decimal",
+    "Date",
+    "Binary",
+}
+
 
 @router.get("/")
 @inject
@@ -139,13 +158,23 @@ async def validate_columns(
 
             column_names = minimal_dataset.column_names
 
-            if len(params.inputs_columns + params.outputs_columns) > len(column_names):
+            group_refs = [
+                ref for ref in (params.input_refs or []) if ref.kind == "group"
+            ]
+
+            if not group_refs and len(
+                params.inputs_columns + params.outputs_columns
+            ) > len(column_names):
                 raise HTTPException(
                     status_code=status.HTTP_400_BAD_REQUEST,
                     detail="Column index out of range",
                 )
 
-            inputs_names = params.inputs_columns
+            inputs_names = (
+                [r.name for r in params.input_refs if r.kind == "raw"]
+                if params.input_refs
+                else params.inputs_columns
+            )
             outputs_names = params.outputs_columns
 
         except exc.SQLAlchemyError as e:
@@ -162,6 +191,33 @@ async def validate_columns(
         )
 
     task: "BaseTask" = component_registry[params.task_name]["class"]()
+
+    if group_refs:
+        declared_types = params.converter_output_types or {}
+        task_metadata = task.get_metadata()
+        allowed_input_types = set(task_metadata.get("inputs_types", []))
+        for ref in group_refs:
+            # A step with a heterogeneous scope (see SessionPreprocessor.
+            # _classify_by_type) can declare more than one type, one per
+            # slot — "{step}:{slot}" disambiguates which one a slotted ref
+            # means; an unslotted ref (the whole step) keeps the plain
+            # "{step}" key, unchanged from before slots existed.
+            key = str(ref.step) if ref.slot is None else f"{ref.step}:{ref.slot}"
+            declared_type = declared_types.get(key)
+            type_ok = declared_type in allowed_input_types or (
+                "DashAIValue" in allowed_input_types
+                and declared_type in _DASHAI_VALUE_TYPE_NAMES
+            )
+            if allowed_input_types and not type_ok:
+                return {
+                    "dataset_status": "invalid",
+                    "error": (
+                        f"Converter step {ref.step} declares output type "
+                        f"'{declared_type}', which is not one of the task's "
+                        f"allowed input types {sorted(allowed_input_types)}."
+                    ),
+                }
+
     validation_response = {}
 
     try:
@@ -236,6 +292,7 @@ async def create_model_session(
     params: ModelSessionParams,
     session_factory: "sessionmaker" = Depends(lambda: di["session_factory"]),
     component_registry: "ComponentRegistry" = Depends(lambda: di["component_registry"]),
+    job_queue: "BaseJobQueue" = Depends(lambda: di["job_queue"]),
 ):
     """Create a new model session.
 
@@ -268,6 +325,28 @@ async def create_model_session(
 
     _validate_splits(params.splits, component_registry)
 
+    sequence = ConverterSequence(steps=params.preprocessing)
+    try:
+        sequence.validate_scopes()
+    except ValueError as e:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=str(e),
+        ) from e
+    has_preprocessing = len(sequence.steps) > 0
+
+    if has_preprocessing and not params.input_column_refs:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=(
+                "input_column_refs is required when preprocessing steps are provided"
+            ),
+        )
+
+    input_column_refs = params.input_column_refs or [
+        RawColumnRef(name=name) for name in params.input_columns
+    ]
+
     with session_factory() as db:
         try:
             dataset = db.get(Dataset, params.dataset_id)
@@ -283,7 +362,12 @@ async def create_model_session(
                 schema = reader.schema
                 column_names = schema.names
 
-            if len(params.input_columns + params.output_columns) > len(column_names):
+            # When preprocessing is configured, input_columns is only a
+            # placeholder until PreprocessingJob resolves the real ones, so
+            # it cannot be checked against the raw dataset's column count.
+            if not has_preprocessing and len(
+                params.input_columns + params.output_columns
+            ) > len(column_names):
                 raise HTTPException(
                     status_code=status.HTTP_400_BAD_REQUEST,
                     detail="Column index out of range",
@@ -300,10 +384,24 @@ async def create_model_session(
                 test_metrics=params.test_metrics,
                 evaluation_strategy=params.evaluation_strategy,
                 splits=params.splits,
+                preprocessing=sequence.model_dump(mode="json"),
+                input_column_refs=[
+                    ref.model_dump(mode="json") for ref in input_column_refs
+                ],
+                preprocessing_status="pending" if has_preprocessing else "ready",
             )
             db.add(model_session)
             db.commit()
             db.refresh(model_session)
+
+            if has_preprocessing:
+                job_id = job_queue.put(
+                    PreprocessingJob(model_session_id=model_session.id)
+                ).id
+                model_session.preprocessing_job_id = str(job_id)
+                db.commit()
+                db.refresh(model_session)
+
             return model_session
         except exc.IntegrityError as e:
             db.rollback()

DashAI/back/api/api_v1/endpoints/runs.py

@@ -295,6 +295,19 @@ async def upload_run(
                     status_code=status.HTTP_404_NOT_FOUND,
                     detail="Model session not found",
                 )
+            if model_session.preprocessing_status == "pending":
+                raise HTTPException(
+                    status_code=status.HTTP_409_CONFLICT,
+                    detail="This session's preprocessing has not finished yet.",
+                )
+            if model_session.preprocessing_status == "failed":
+                raise HTTPException(
+                    status_code=status.HTTP_409_CONFLICT,
+                    detail=(
+                        "This session's preprocessing failed: "
+                        f"{model_session.preprocessing_error}"
+                    ),
+                )
             # REQUIRES_DOWNLOAD is the static contract; the download state is
             # reconciled against the filesystem so a model downloaded after
             # startup (in the worker process) is recognised without a restart.

DashAI/back/api/api_v1/schemas/model_sessions_params.py

@@ -1,6 +1,8 @@
-from typing import List
+from typing import Dict, List, Optional
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
+
+from DashAI.back.preprocessing.column_ref import ColumnRef, ConverterStep
 
 
 class ModelSessionParams(BaseModel):
@@ -14,13 +16,17 @@ class ModelSessionParams(BaseModel):
     test_metrics: List[str]
     evaluation_strategy: str
     splits: str
+    preprocessing: List[ConverterStep] = Field(default_factory=list)
+    input_column_refs: Optional[List[ColumnRef]] = None
 
 
 class ColumnsValidationParams(BaseModel):
     task_name: str
     dataset_id: int
     inputs_columns: List[str]
     outputs_columns: List[str]
+    input_refs: Optional[List[ColumnRef]] = None
+    converter_output_types: Optional[Dict[str, str]] = None
 
 
 class ModelSessionBulkDeleteParams(BaseModel):

DashAI/back/config.py

@@ -19,6 +19,7 @@ class DefaultSettings(BaseSettings):
     DATASETS_PATH: str = "datasets"
     IMAGES_PATH: str = "images"
     RUNS_PATH: str = "runs"
+    PREPROCESSING_PATH: str = "preprocessing"
     EXPLANATIONS_PATH: str = "explanations"
     EXPLORATIONS_PATH: str = "explorations"
     DOCUMENTS_PATH: str = "documents"

DashAI/back/converters/base_converter.py

@@ -42,6 +42,15 @@ class BaseConverter(ConfigObject, ABC):
     COLOR: Final[str] = "rgb(255, 255, 255)"
     SUPERVISED: bool = False
     CHANGES_ROW_COUNT: bool = False
+    # True for converters that never transform values, only keep or drop
+    # whole columns as-is (feature selection, variance thresholding): the
+    # output type of a surviving column is always exactly its input type, no
+    # arithmetic involved. Lets a caller that already knows the real input
+    # type (e.g. the Models-module wizard, once a real scope is chosen) use
+    # that instead of this class's own best-effort get_output_type() guess,
+    # which — called on a bare unfitted instance — has no idea what column
+    # it will actually run on.
+    PRESERVES_INPUT_TYPE: bool = False
     SCHEMA: BaseConverterSchema
 
     @classmethod
@@ -74,6 +83,7 @@ def get_metadata(cls) -> Dict[str, Any]:
         meta["download_size_bytes"] = getattr(cls, "DOWNLOAD_SIZE_BYTES", None)
         meta["supervised"] = cls.SUPERVISED
         meta["changes_row_count"] = cls.CHANGES_ROW_COUNT
+        meta["preserves_input_type"] = cls.PRESERVES_INPUT_TYPE
         meta["n_components_features_bounded"] = getattr(
             cls, "N_COMPONENTS_FEATURES_BOUNDED", False
         )
@@ -108,6 +118,30 @@ def get_metadata(cls) -> Dict[str, Any]:
         # Drop restricted_dtypes (no converter uses it; it is always [])
         meta.pop("restricted_dtypes", None)
 
+        # A representative output type, so the Models-module wizard can show
+        # "this converter's group is typed X" before any real fit exists.
+        # Not every converter can be instantiated with no arguments (some
+        # require constructor params with no default), so this is
+        # best-effort: None means "unknown until configured".
+        try:
+            output_type = cls().get_output_type()
+            meta["output_type"] = (
+                output_type.display_name()
+                if output_type is not None and hasattr(output_type, "display_name")
+                else None
+            )
+            # The concrete storage dtype (e.g. "int64"), so a group column can
+            # show one instead of "unknown" before any real fit exists — same
+            # best-effort default-constructed instance as output_type above.
+            meta["output_dtype"] = (
+                output_type.to_string().get("dtype")
+                if output_type is not None and hasattr(output_type, "to_string")
+                else None
+            )
+        except Exception:
+            meta["output_type"] = None
+            meta["output_dtype"] = None
+
         return meta
 
     @abstractmethod

DashAI/back/converters/category/feature_selection.py

@@ -34,6 +34,7 @@ class FeatureSelectionConverter(BaseConverter):
     )
     ICON: Final[str] = Icon.FilterList.value
     COLOR: Final[str] = "rgb(255, 206, 86)"
+    PRESERVES_INPUT_TYPE = True
 
     def fit(
         self, x: "DashAIDataset", y: Union["DashAIDataset", None] = None