Make stage execution typed end to end

modules/core/src/main/scala/com/flowforge/core/PipelineBuilder.scala

@@ -3,7 +3,7 @@ package com.flowforge.core
 import cats.data.Kleisli
 import com.flowforge.core.algebra.EffectSystem
 import com.flowforge.core.contracts.{ SchemaConforms, SchemaPolicy }
-import com.flowforge.core.exec.{ ExecutableStage, StageComposer }
+import com.flowforge.core.exec.{ StageChain, StageComposer, StageKind }
 import com.flowforge.core.lineage.OpenLineageEmitter
 import com.flowforge.core.observability.Tracer
 import com.flowforge.core.types.BuilderState.{ WithContract, WithTransform }
@@ -32,7 +32,7 @@ import com.flowforge.framework.{ Pipeline => FFPipeline, PipelineMetadata }
 case class PipelineBuilder[S <: BuilderState, F[_]: EffectSystem, In, Out] private (
   name: String,
   description: String = "",
-  stages: List[PipelineStage[F, _, _]] = List.empty,
+  stages: StageChain[F, In, Out],
   config: Option[PipelineConfig] = None,
   lineageEmitter: Option[OpenLineageEmitter[F]] = None,
   tracer: Option[Tracer[F]] = None) {
@@ -53,18 +53,23 @@ case class PipelineBuilder[S <: BuilderState, F[_]: EffectSystem, In, Out] priva
   /**
    * Append a stage and move to the next phantom state.
    *
-   * `copy` cannot do this because the phantom state and the In/Out types change, and `copy` returns the same
+   * `copy` cannot do this because the phantom state and the Out type change, and `copy` returns the same
    * type. That forced a hand-written constructor call in each stage method, and three of them listed five of
    * the six fields, so a tracer attached before the first stage was dropped. Carrying the fields here means
    * there is one place to update when a field is added.
+   *
+   * The arrow has to read what the builder currently produces, which is what makes the stage chain and the
+   * builder's own type parameters say the same thing rather than two things that happen to agree.
    */
-  private def advance[S2 <: BuilderState, In2, Out2](
-    stage: PipelineStage[F, _, _],
-  ): PipelineBuilder[S2, F, In2, Out2] =
-    PipelineBuilder[S2, F, In2, Out2](
+  private def advance[S2 <: BuilderState, Out2](
+    stageName: String,
+    kind: StageKind,
+    arrow: Kleisli[F, Out, Out2],
+  ): PipelineBuilder[S2, F, In, Out2] =
+    PipelineBuilder[S2, F, In, Out2](
       name,
       description,
-      stages :+ stage,
+      stages.andThen(stageName, kind, arrow),
       config,
       lineageEmitter,
       tracer,
@@ -76,19 +81,27 @@ case class PipelineBuilder[S <: BuilderState, F[_]: EffectSystem, In, Out] priva
    *
    * SOURCE: produced C must conform to declared contract R under policy P Advances phantom state: Empty ->
    * HasSource with HasContract
+   *
+   * A source reads from outside the pipeline, so its arrow takes no input. The `Out =:= Unit` evidence
+   * restricts it to a builder that has not produced a value yet; without it a source could be appended after
+   * a transform, which discarded the transform's output. At `Empty` the builder's `Out` is already `Unit`, so
+   * the evidence resolves on its own and no caller names it.
    */
   def addTypedSource[C, R, P <: SchemaPolicy](
     source: TypedSource[R],
     reader: DataSource => F[C],
-  )(implicit ev: SchemaConforms[C, R, P],
-  ): PipelineBuilder[WithContract, F, Unit, C] = {
+  )(implicit
+    ev: SchemaConforms[C, R, P],
+    atStart: Out =:= Unit,
+  ): PipelineBuilder[WithContract, F, In, C] = {
     val stage = PipelineStage.Source[F, C](
       name = s"contract-source-${stages.size}",
       description = s"Contract-aware source with compile-time validation",
       dataSource = source.underlying,
       execute = Kleisli(_ => reader(source.underlying)),
     )
-    advance[WithContract, Unit, C](stage)
+    // `=:=` is a function in both Scala versions, so the source's Unit input is adapted rather than cast.
+    advance[WithContract, C](stage.name, StageKind.Source, stage.execute.local[Out](atStart))
   }
 
   /**
@@ -105,7 +118,7 @@ case class PipelineBuilder[S <: BuilderState, F[_]: EffectSystem, In, Out] priva
       description = "Contract-aware transformation",
       execute = Kleisli(transform),
     )
-    advance[WithTransform, In, C](stage)
+    advance[WithTransform, C](stage.name, StageKind.Transform, stage.execute)
   }
 
   /**
@@ -148,7 +161,7 @@ case class PipelineBuilder[S <: BuilderState, F[_]: EffectSystem, In, Out] priva
         writer(data, sink.underlying),
       ),
     )
-    advance[BuilderState.Complete, In, Unit](stage)
+    advance[BuilderState.Complete, Unit](stage.name, StageKind.Sink, stage.execute)
   }
 
   /**
@@ -162,15 +175,15 @@ case class PipelineBuilder[S <: BuilderState, F[_]: EffectSystem, In, Out] priva
 
     val kleisli: Kleisli[F, In, Out] = StageComposer.compose[F, In, Out](
       pipelineName = name,
-      stages = stages.map(st => ExecutableStage[F](st.name, st.execute)),
+      stages = stages,
       tracer = tracer,
       lineage = lineageEmitter,
     )
 
     val md = PipelineMetadata(
       name = name,
-      stages = stages.map(_.name),
-      transformations = stages.count(_.isTransform),
+      stages = stages.names,
+      transformations = stages.count(StageKind.Transform),
       qualityChecks = 0,
       tags = Map(
         "builder" -> "contract-aware",
@@ -191,10 +204,10 @@ object PipelineBuilder {
    * build.
    */
   def apply[F[_]: EffectSystem](name: String): PipelineBuilder[BuilderState.Empty, F, Unit, Unit] =
-    // `stages` is passed rather than defaulted: Scala 3 solves a default argument's type parameters on
-    // their own, so the default would come back as List[PipelineStage[Nothing, _, _]].
+    // `stages` has no default: an empty chain's input and output types are the same type, which is only true
+    // of a builder that has not added a stage yet, so there is nothing for a default to mean elsewhere.
     PipelineBuilder[BuilderState.Empty, F, Unit, Unit](
       name,
-      stages = List.empty[PipelineStage[F, _, _]],
+      stages = StageChain.empty[F, Unit],
     )
 }

modules/core/src/main/scala/com/flowforge/core/exec/StageChain.scala

@@ -0,0 +1,157 @@
+package com.flowforge.core.exec
+
+import cats.data.Kleisli
+import com.flowforge.core.algebra.EffectSystem
+
+/**
+ * What a stage is for, as far as anything outside the stage can tell.
+ *
+ * Every builder reported its own stage counts its own way: one read a `isTransform` field, two used
+ * `isInstanceOf`, two used `collect`, and two just reported the stage count. Five answers to one question is
+ * five chances to disagree. A builder now says what kind a stage is when it appends it, and the counts come
+ * from that.
+ */
+sealed abstract class StageKind extends Product with Serializable
+
+object StageKind {
+
+  /** Reads from outside the pipeline, so it takes no input. */
+  case object Source extends StageKind
+
+  /** Turns the previous stage's output into the next stage's input. */
+  case object Transform extends StageKind
+
+  /** Checks the data against a contract and passes it through. */
+  case object Quality extends StageKind
+
+  /** Writes to outside the pipeline, so it produces no value. */
+  case object Sink extends StageKind
+}
+
+/**
+ * The stages of a pipeline, in order, holding the type chain that runs through them.
+ *
+ * A `List[Kleisli[F, ?, ?]]` cannot hold this. A list has one element type, and a pipeline's stages do not:
+ * stage one is `A => B` and stage two is `B => C`. The old composer erased every stage to
+ * `Kleisli[F, Any, Any]`, folded the list, and cast the result back to the types the builder claimed. Erasure
+ * makes that cast silent, so a builder whose types did not line up produced an arrow that handed one stage's
+ * output to a stage expecting something else, with no cast error to say so.
+ *
+ * This type is the same sequence with the links kept. `Link` holds the type between two stages as its own
+ * parameter, so appending a stage that does not read the previous stage's output does not compile and there
+ * is nothing left to cast.
+ *
+ * The trade: `Mid` is hidden inside `Link`, so the chain can be folded but not re-indexed or reordered after
+ * it is built. That is what a type-aligned sequence costs in Scala 2.13, and nothing here reorders stages
+ * after building.
+ */
+sealed abstract class StageChain[F[_], In, Out] extends Product with Serializable {
+
+  /**
+   * Each stage's name and kind, most recently appended first.
+   *
+   * Reversed because that is the order a cons list can be extended in without copying. Appending with `:+`
+   * copies every node before it, so a chain of n stages would hold n lists of average length n/2 rather than
+   * sharing one spine, and a pipeline whose stages are generated rather than written out would pay for it.
+   * Only `names` needs the running order, and only once per build.
+   */
+  private[exec] def entriesReversed: List[StageChain.Entry]
+
+  /** How many stages the chain holds. Held per link, because a builder reads it on every append. */
+  def size: Int
+
+  /** The stage names, in the order they run. */
+  final def names: List[String] = entriesReversed.reverseIterator.map(_.name).toList
+
+  /** How many stages of one kind the chain holds. */
+  final def count(kind: StageKind): Int = entriesReversed.count(_.kind == kind)
+
+  /** Append a stage. It has to read what the chain currently produces. */
+  final def andThen[Next](
+    stageName: String,
+    kind: StageKind,
+    stage: Kleisli[F, Out, Next],
+  ): StageChain[F, In, Next] =
+    StageChain.Link(this, StageChain.Entry(stageName, kind), stage)
+
+  /**
+   * The arrow that runs the whole chain, with `decorate` applied to each stage.
+   *
+   * `decorate` is applied at each stage's own input and output types rather than at `Any`, which is what
+   * removes the need to erase. See [[StageDecorator]].
+   *
+   * Returning the arrow costs one call per chain, not one stack frame per stage. Each link defers the rest of
+   * the chain through `EffectSystem.suspend`, so the descent happens inside `F`'s own loop rather than on the
+   * JVM stack. Walking the links eagerly instead would bound the chain's depth by whatever stack the thread
+   * happens to have, which is not a property of the pipeline: a chain that ran on a 8MB main thread
+   * overflowed at a third of the depth on a 1MB CI runner. The cost is that the links are walked once per run
+   * rather than once per build, and that stack safety rests on `F.flatMap` and `F.suspend` being stack-safe.
+   * `MonadError` does not promise that, but every effect type worth running a pipeline in provides it.
+   * [[StageChainSpec]] runs a chain deep enough that an eager walk would fail.
+   */
+  private[exec] def arrow(decorate: StageDecorator[F])(implicit F: EffectSystem[F]): Kleisli[F, In, Out]
+}
+
+object StageChain {
+
+  /** A chain with no stages. Its arrow is the identity, which is the honest answer for no stages. */
+  def empty[F[_], A]: StageChain[F, A, A] = Identity[F, A]()
+
+  private[exec] final case class Entry(name: String, kind: StageKind)
+
+  /**
+   * `arrow` is an abstract method each case overrides rather than a match on the two cases.
+   *
+   * A match would have to recover `In =:= Out` for `Identity` and the hidden `Mid` for `Link` from the
+   * scrutinee's type, which is GADT refinement. Scala 2.13 does that unreliably. Inside a case's own body its
+   * type parameters are already in scope, so there is nothing to infer and the same source compiles on 2.13
+   * and 3.3.
+   */
+  private final case class Identity[F[_], A]() extends StageChain[F, A, A] {
+
+    private[exec] def entriesReversed: List[Entry] = Nil
+
+    val size: Int = 0
+
+    private[exec] def arrow(
+      decorate: StageDecorator[F],
+    )(implicit F: EffectSystem[F],
+    ): Kleisli[F, A, A] = Kleisli((a: A) => F.pure(a))
+  }
+
+  private final case class Link[F[_], In, Mid, Out](
+    init: StageChain[F, In, Mid],
+    entry: Entry,
+    stage: Kleisli[F, Mid, Out])
+      extends StageChain[F, In, Out] {
+
+    // Both are computed once per append and share the preceding link's spine, so appending is constant
+    // time and the whole chain holds one list. The next stage's name is derived from `size`, so a builder
+    // reads it on every append.
+    private[exec] val entriesReversed: List[Entry] = entry :: init.entriesReversed
+
+    val size: Int = init.size + 1
+
+    private[exec] def arrow(
+      decorate: StageDecorator[F],
+    )(implicit F: EffectSystem[F],
+    ): Kleisli[F, In, Out] = {
+      val head = decorate(entry.name, stage)
+      // `suspend` is what keeps the preceding links off the stack: the rest of the chain is built when `F`
+      // reaches this step, not when the arrow is handed out, so the depth is `F`'s to carry rather than the
+      // thread's. `andThen` here would build the whole chain eagerly and recurse once per stage.
+      Kleisli[F, In, Out](in => F.flatMap(F.suspend(init.arrow(decorate).run(in)))(head.run))
+    }
+  }
+}
+
+/**
+ * Per-stage wrapping, applied at the stage's own types.
+ *
+ * The type parameters are on `apply` rather than on the trait on purpose. A decorator typed
+ * `Kleisli[F, Any, Any] => Kleisli[F, Any, Any]` is what forced the composer to erase its stages in the first
+ * place. One that can be applied at whatever types a stage happens to have does not.
+ */
+private[exec] trait StageDecorator[F[_]] {
+  def apply[A, B](stageName: String, stage: Kleisli[F, A, B]): Kleisli[F, A, B]
+}