One implementation of model training: composable units, the engine that runs them, and the cross-validation reconciliation

DashAI/alembic/versions/82fb7a6b8ac2_merge_pipeline_tracking_and_cross_.py

@@ -0,0 +1,37 @@
+"""Merge the pipeline tracking and cross-validation heads.
+
+Two lines of work reached the database at the same time and neither knew about
+the other, so the revision graph ended with two heads and ``upgrade head``
+refused to pick one:
+
+``e7b4d1a9c206`` added the pipeline run tracking tables (``pipeline_run``,
+``pipeline_node_run``, ``pipeline_node_artifact``); ``b7e4d2a19c63`` is the tip
+of the branch that brought cross-validation and the evaluation reports, which
+added ``model_session.evaluation_strategy``, ``run.nested``, the fold columns on
+``metric`` and the ``report`` table.
+
+There is nothing to do here. The two sides touch disjoint tables and disjoint
+columns, so joining the graph is the whole of the change: this revision exists
+to give the two heads a single successor.
+
+Revision ID: 82fb7a6b8ac2
+Revises: b7e4d2a19c63, e7b4d1a9c206
+Create Date: 2026-09-10 13:03:45.322840
+
+"""
+
+from typing import Sequence, Union
+
+# revision identifiers, used by Alembic.
+revision: str = "82fb7a6b8ac2"
+down_revision: Union[str, None] = ("b7e4d2a19c63", "e7b4d1a9c206")
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Join the two revision lines. Neither side needs anything applied."""
+
+
+def downgrade() -> None:
+    """Split them again, which is likewise nothing to undo."""

DashAI/alembic/versions/e7b4d1a9c206_add_pipeline_run_tracking.py

@@ -0,0 +1,122 @@
+"""Add pipeline run tracking tables.
+
+Separates the definition of a graph from its executions. ``Pipeline`` keeps the
+definition, which changes as the user edits it; ``pipeline_run`` freezes the
+steps and edges of one execution, ``pipeline_node_run`` tracks each node of it,
+and ``pipeline_node_artifact`` holds what a node emitted, keyed by a key from
+the unit's PROVIDES rather than by a column per node type.
+
+The three JSON result columns on ``pipeline`` (exploration, train, prediction)
+are deliberately left in place: the pipelines endpoints and the front's results
+view still read them, so removing them belongs with rewriting those.
+
+Revision ID: e7b4d1a9c206
+Revises: d5b3c8f2a041
+Create Date: 2026-08-20 00:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "e7b4d1a9c206"
+down_revision: Union[str, None] = "d5b3c8f2a041"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "pipeline_run",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("pipeline_id", sa.Integer(), nullable=False),
+        sa.Column("steps", sa.JSON(), nullable=True),
+        sa.Column("edges", sa.JSON(), nullable=True),
+        sa.Column("created", sa.DateTime(), nullable=False),
+        sa.Column("last_modified", sa.DateTime(), nullable=False),
+        sa.Column("delivery_time", sa.DateTime(), nullable=True),
+        sa.Column("start_time", sa.DateTime(), nullable=True),
+        sa.Column("end_time", sa.DateTime(), nullable=True),
+        sa.Column(
+            "status",
+            sa.Enum(
+                "NOT_STARTED",
+                "DELIVERED",
+                "STARTED",
+                "FINISHED",
+                "ERROR",
+                name="pipelinerunstatus",
+            ),
+            nullable=False,
+        ),
+        sa.Column("error_message", sa.String(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["pipeline_id"],
+            ["pipeline.id"],
+            name=op.f("fk_pipeline_run_pipeline_id_pipeline"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_pipeline_run")),
+    )
+
+    op.create_table(
+        "pipeline_node_run",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("pipeline_run_id", sa.Integer(), nullable=False),
+        sa.Column("node_id", sa.String(), nullable=False),
+        sa.Column("block_id", sa.String(), nullable=False),
+        sa.Column("node_type", sa.String(), nullable=False),
+        sa.Column("config", sa.JSON(), nullable=True),
+        sa.Column("input", sa.JSON(), nullable=True),
+        sa.Column("output", sa.JSON(), nullable=True),
+        sa.Column("created", sa.DateTime(), nullable=False),
+        sa.Column("last_modified", sa.DateTime(), nullable=False),
+        sa.Column("delivery_time", sa.DateTime(), nullable=True),
+        sa.Column("start_time", sa.DateTime(), nullable=True),
+        sa.Column("end_time", sa.DateTime(), nullable=True),
+        sa.Column(
+            "status",
+            sa.Enum(
+                "NOT_STARTED",
+                "DELIVERED",
+                "STARTED",
+                "FINISHED",
+                "ERROR",
+                "CANCELLED",
+                name="noderunstatus",
+            ),
+            nullable=False,
+        ),
+        sa.Column("error_message", sa.String(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["pipeline_run_id"],
+            ["pipeline_run.id"],
+            name=op.f("fk_pipeline_node_run_pipeline_run_id_pipeline_run"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_pipeline_node_run")),
+    )
+
+    op.create_table(
+        "pipeline_node_artifact",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("node_run_id", sa.Integer(), nullable=False),
+        sa.Column("key", sa.String(), nullable=False),
+        sa.Column("value", sa.JSON(), nullable=True),
+        sa.Column("created", sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["node_run_id"],
+            ["pipeline_node_run.id"],
+            name=op.f("fk_pipeline_node_artifact_node_run_id_pipeline_node_run"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_pipeline_node_artifact")),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("pipeline_node_artifact")
+    op.drop_table("pipeline_node_run")
+    op.drop_table("pipeline_run")

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

@@ -330,7 +330,6 @@ async def delete_prediction(
 @inject
 async def preview_manual_prediction(
     request: Request,
-    component_registry: "ComponentRegistry" = Depends(lambda: di["component_registry"]),
     session_factory: "sessionmaker" = Depends(lambda: di["session_factory"]),
 ):
     """Run a synchronous manual prediction and return results without persisting.
@@ -413,7 +412,6 @@ async def preview_manual_prediction(
         run_manual_prediction,
         run_id=run_id_int,
         manual_input_data=rows_data,
-        component_registry=component_registry,
         session_factory=session_factory,
     )
     return {"columns": columns, "rows": rows}

DashAI/back/core/enums/status.py

@@ -67,3 +67,24 @@ class DatafileStatus(Enum):
     DOWNLOADING = "downloading"
     READY = "ready"
     ERROR = "error"
+
+
+class PipelineRunStatus(Enum):
+    NOT_STARTED = 0
+    DELIVERED = 1
+    STARTED = 2
+    FINISHED = 3
+    ERROR = 4
+
+
+class NodeRunStatus(Enum):
+    NOT_STARTED = 0
+    DELIVERED = 1
+    STARTED = 2
+    FINISHED = 3
+    ERROR = 4
+    # A node that never ran because an earlier one failed. Distinct from
+    # NOT_STARTED, which is a node still waiting its turn: without the
+    # distinction a run that died halfway is indistinguishable from one still
+    # in flight.
+    CANCELLED = 5

DashAI/back/dag/__init__.py

@@ -0,0 +1,28 @@
+"""A sequential DAG engine over the atomic units.
+
+Nothing here imports from ``DashAI/back/pipeline/``. That subsystem does not
+run — its nodes implement two of ``BaseJob``'s four abstract methods, so they
+cannot even be instantiated — and it is replaced rather than repaired.
+
+The engine is deliberately sequential. Its predecessor was concurrent, and the
+concurrency was not buying what it cost: the most expensive node in a pipeline
+was already serialised behind an exclusive lock, every context and database
+write was wrapped in another, and SQLite answered the remaining concurrent
+writes with "database is locked" often enough to need retries with exponential
+backoff. Units open a database session each, which is safe in sequence and was
+the source of that contention in parallel.
+"""
+
+from DashAI.back.dag.graph import Edge, Graph, GraphError, Node, connect, sinks
+from DashAI.back.dag.validate import resolve_unit_class, validate
+
+__all__ = [
+    "Edge",
+    "Graph",
+    "GraphError",
+    "Node",
+    "connect",
+    "resolve_unit_class",
+    "sinks",
+    "validate",
+]