From 49f1c008868bc3b4ccf0ef830efb7bd7e055c1fc Mon Sep 17 00:00:00 2001
From: Vitthal Mirji <vitthalmirji@gmail.com>
Date: Tue, 6 Oct 2026 17:07:30 +0200
Subject: [PATCH 1/4] feat(core): add a stage list that holds the type between
 stages

---
 .../com/flowforge/core/exec/StageChain.scala  | 130 ++++++++++++++++++
 .../flowforge/core/exec/StageChainSpec.scala  |  60 ++++++++
 2 files changed, 190 insertions(+)
 create mode 100644 modules/core/src/main/scala/com/flowforge/core/exec/StageChain.scala
 create mode 100644 modules/core/src/test/scala/com/flowforge/core/exec/StageChainSpec.scala

diff --git a/modules/core/src/main/scala/com/flowforge/core/exec/StageChain.scala b/modules/core/src/main/scala/com/flowforge/core/exec/StageChain.scala
new file mode 100644
index 00000000..c53d9c2f
--- /dev/null
+++ b/modules/core/src/main/scala/com/flowforge/core/exec/StageChain.scala
@@ -0,0 +1,130 @@
+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, in the order they run. */
+  private[exec] def entries: List[StageChain.Entry]
+
+  /** The stage names, in the order they run. */
+  final def names: List[String] = entries.map(_.name)
+
+  /** How many stages the chain holds. */
+  final def size: Int = entries.size
+
+  /** How many stages of one kind the chain holds. */
+  final def count(kind: StageKind): Int = entries.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]].
+   */
+  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 entries: List[Entry] = Nil
+
+    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] {
+
+    // Computed once per append rather than per read: `names`, `size` and `count` are all called while a
+    // pipeline is being built, and the next stage's name is derived from `size`.
+    private[exec] val entries: List[Entry] = init.entries :+ entry
+
+    private[exec] def arrow(
+      decorate: StageDecorator[F],
+    )(implicit F: EffectSystem[F],
+    ): Kleisli[F, In, Out] = init.arrow(decorate).andThen(decorate(entry.name, stage))
+  }
+}
+
+/**
+ * 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]
+}
diff --git a/modules/core/src/test/scala/com/flowforge/core/exec/StageChainSpec.scala b/modules/core/src/test/scala/com/flowforge/core/exec/StageChainSpec.scala
new file mode 100644
index 00000000..047af58b
--- /dev/null
+++ b/modules/core/src/test/scala/com/flowforge/core/exec/StageChainSpec.scala
@@ -0,0 +1,60 @@
+package com.flowforge.core.exec
+
+import cats.data.Kleisli
+import cats.effect.IO
+import cats.effect.unsafe.implicits.global
+import com.flowforge.core.algebra.EffectSystem
+import com.flowforge.core.instances.EffectInstances
+import org.scalatest.funsuite.AnyFunSuite
+import org.scalatest.matchers.should.Matchers
+
+/**
+ * Covers what the chain exists to provide: the type between two stages is checked when the second one is
+ * appended, so the composer has nothing left to cast.
+ */
+class StageChainSpec extends AnyFunSuite with Matchers {
+
+  implicit val es: EffectSystem[IO] = EffectInstances.catsEffectSystemInstance
+
+  /** The decorator the composer would supply, with its wrapping left out. */
+  private val plain: StageDecorator[IO] = new StageDecorator[IO] {
+    def apply[A, B](stageName: String, stage: Kleisli[IO, A, B]): Kleisli[IO, A, B] = stage
+  }
+
+  private def chain: StageChain[IO, Int, String] =
+    StageChain
+      .empty[IO, Int]
+      .andThen("double", StageKind.Transform, Kleisli((n: Int) => IO.pure(n * 2)))
+      .andThen("render", StageKind.Transform, Kleisli((n: Int) => IO.pure(s"n=$n")))
+
+  test("an empty chain runs as the identity") {
+    StageChain.empty[IO, Int].arrow(plain).run(7).unsafeRunSync() shouldBe 7
+  }
+
+  test("the arrow runs the stages in the order they were appended") {
+    chain.arrow(plain).run(4).unsafeRunSync() shouldBe "n=8"
+  }
+
+  test("names come back in the order the stages run") {
+    chain.names shouldBe List("double", "render")
+  }
+
+  test("counts come from the kind the builder gave each stage") {
+    val withSink = chain.andThen("write", StageKind.Sink, Kleisli((_: String) => IO.unit))
+    withSink.count(StageKind.Transform) shouldBe 2
+    withSink.count(StageKind.Sink) shouldBe 1
+    withSink.count(StageKind.Quality) shouldBe 0
+    withSink.size shouldBe 3
+  }
+
+  test("a stage that does not read the previous stage's output is rejected") {
+    assertDoesNotCompile(
+      """
+      StageChain
+        .empty[IO, Int]
+        .andThen("double", StageKind.Transform, Kleisli((n: Int) => IO.pure(n * 2)))
+        .andThen("shout", StageKind.Transform, Kleisli((s: String) => IO.pure(s.toUpperCase)))
+      """,
+    )
+  }
+}

From af1d8bc43e78817185561dc03c55ae6540441200 Mon Sep 17 00:00:00 2001
From: Vitthal Mirji <vitthalmirji@gmail.com>
Date: Tue, 6 Oct 2026 17:07:30 +0200
Subject: [PATCH 2/4] fix(core): build the pipeline arrow from the typed stage
 chain, dropping the Any casts

---
 .../com/flowforge/core/PipelineBuilder.scala  |  51 +++--
 .../flowforge/core/exec/StageComposer.scala   |  87 +++-----
 .../core/impl/InMemoryPipelineBuilder.scala   |  66 +++---
 .../engines/spark/SparkPipelineBuilder.scala  | 196 ++++++++++--------
 4 files changed, 219 insertions(+), 181 deletions(-)

diff --git a/modules/core/src/main/scala/com/flowforge/core/PipelineBuilder.scala b/modules/core/src/main/scala/com/flowforge/core/PipelineBuilder.scala
index 8ba4ee4a..219ec99b 100644
--- a/modules/core/src/main/scala/com/flowforge/core/PipelineBuilder.scala
+++ b/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],
     )
 }
diff --git a/modules/core/src/main/scala/com/flowforge/core/exec/StageComposer.scala b/modules/core/src/main/scala/com/flowforge/core/exec/StageComposer.scala
index c9625896..0126d270 100644
--- a/modules/core/src/main/scala/com/flowforge/core/exec/StageComposer.scala
+++ b/modules/core/src/main/scala/com/flowforge/core/exec/StageComposer.scala
@@ -5,34 +5,16 @@ import com.flowforge.core.algebra.EffectSystem
 import com.flowforge.core.lineage.{ LineageError, OpenLineageEmitter }
 import com.flowforge.core.observability.Tracer
 
-/**
- * A stage reduced to the only two things execution needs: a name, and an arrow to run.
- *
- * Builders hold richer stage types (`PipelineStage`, `InMemoryStage`). Those carry sources, sinks and
- * contracts, none of which execution looks at. Narrowing to this type is what lets every builder share one
- * composer instead of each writing its own fold.
- */
-final class ExecutableStage[F[_]] private (val name: String, val run: Kleisli[F, Any, Any])
-
-object ExecutableStage {
-
-  /**
-   * Erase a stage arrow to `Any => Any`.
-   *
-   * The cast is safe because a builder only ever appends a stage whose input type is the previous stage's
-   * output type, and it is unchecked because that chaining is tracked in the builder's type parameters rather
-   * than in the stage list. Keeping the cast here means it exists once rather than once per builder.
-   */
-  def apply[F[_]](name: String, stage: Kleisli[F, _, _]): ExecutableStage[F] =
-    new ExecutableStage[F](name, stage.asInstanceOf[Kleisli[F, Any, Any]])
-}
-
 /**
  * Composes pipeline stages into a single arrow and decorates each stage with tracing and lineage.
  *
- * This is the only place in core that answers "how is a list of stages run". `PipelineBuilder.build` and the
+ * This is the only place in core that answers "how is a chain of stages run". `PipelineBuilder.build` and the
  * in-memory builders each used to answer it separately. They disagreed, and each one was missing something
  * the others had.
+ *
+ * It holds no casts. The stages arrive as a [[StageChain]], which carries the type between each pair of
+ * stages, so the arrow this builds has the caller's `In` and `Out` because the chain does, not because it was
+ * told to.
  */
 object StageComposer {
 
@@ -50,19 +32,17 @@ object StageComposer {
    */
   def compose[F[_], In, Out](
     pipelineName: String,
-    stages: List[ExecutableStage[F]],
+    stages: StageChain[F, In, Out],
     tracer: Option[Tracer[F]] = None,
     lineage: Option[OpenLineageEmitter[F]] = None,
   )(implicit F: EffectSystem[F],
-  ): Kleisli[F, In, Out] = {
-    val erased = Kleisli[F, Any, Any] { input =>
+  ): Kleisli[F, In, Out] =
+    Kleisli[F, In, Out] { input =>
       lineage match {
-        case None          => chain(stages, tracer, None).run(input)
+        case None          => stages.arrow(decorator(tracer, None)).run(input)
         case Some(emitter) => runTracked(pipelineName, stages, tracer, emitter, input)
       }
     }
-    erased.asInstanceOf[Kleisli[F, In, Out]]
-  }
 
   /**
    * Emitter, namespace and the pipeline's own run id, resolved once per execution.
@@ -75,14 +55,14 @@ object StageComposer {
     pipelineRunId: String,
     namespace: String)
 
-  private def runTracked[F[_]](
+  private def runTracked[F[_], In, Out](
     pipelineName: String,
-    stages: List[ExecutableStage[F]],
+    stages: StageChain[F, In, Out],
     tracer: Option[Tracer[F]],
     emitter: OpenLineageEmitter[F],
-    input: Any,
+    input: In,
   )(implicit F: EffectSystem[F],
-  ): F[Any] = {
+  ): F[Out] = {
     // Both reads are effects: generateRunId reads the environment and otherwise draws a random id.
     val context = F.delay(
       RunContext(
@@ -92,20 +72,15 @@ object StageComposer {
       ),
     )
     F.flatMap(context) { ctx =>
-      around(ctx, pipelineName, ctx.pipelineRunId, chain(stages, tracer, Some(ctx)).run(input))
+      around(
+        ctx,
+        pipelineName,
+        ctx.pipelineRunId,
+        stages.arrow(decorator(tracer, Some(ctx))).run(input),
+      )
     }
   }
 
-  private def chain[F[_]](
-    stages: List[ExecutableStage[F]],
-    tracer: Option[Tracer[F]],
-    lineage: Option[RunContext[F]],
-  )(implicit F: EffectSystem[F],
-  ): Kleisli[F, Any, Any] =
-    stages.foldLeft(Kleisli.ask[F, Any]) { (acc, stage) =>
-      acc.andThen(observe(stage, tracer, lineage))
-    }
-
   /**
    * Wrap one stage. The span and the events cover this stage and nothing before it.
    *
@@ -115,20 +90,22 @@ object StageComposer {
    * belongs in a parent run facet, which the current `emitJobStart(namespace, jobName, runId)` signature
    * cannot carry; adding it means changing the emitter interface, which is left out of this fix.
    */
-  private def observe[F[_]](
-    stage: ExecutableStage[F],
+  private def decorator[F[_]](
     tracer: Option[Tracer[F]],
     lineage: Option[RunContext[F]],
   )(implicit F: EffectSystem[F],
-  ): Kleisli[F, Any, Any] =
-    Kleisli[F, Any, Any] { in =>
-      val body   = stage.run.run(in)
-      val traced = tracer.fold(body)(t => t.inSpan(stage.name)(body))
-      lineage.fold(traced) { ctx =>
-        // Random rather than OpenLineageEmitter.generateRunId: an orchestrator supplied id names the
-        // pipeline run, so handing it to a stage would recreate the one-id-many-jobs problem.
-        F.flatMap(F.delay(java.util.UUID.randomUUID().toString))(around(ctx, stage.name, _, traced))
-      }
+  ): StageDecorator[F] =
+    new StageDecorator[F] {
+      def apply[A, B](stageName: String, stage: Kleisli[F, A, B]): Kleisli[F, A, B] =
+        Kleisli[F, A, B] { in =>
+          val body   = stage.run(in)
+          val traced = tracer.fold(body)(t => t.inSpan(stageName)(body))
+          lineage.fold(traced) { ctx =>
+            // Random rather than OpenLineageEmitter.generateRunId: an orchestrator supplied id names the
+            // pipeline run, so handing it to a stage would recreate the one-id-many-jobs problem.
+            F.flatMap(F.delay(java.util.UUID.randomUUID().toString))(around(ctx, stageName, _, traced))
+          }
+        }
     }
 
   /** START before, COMPLETE on success, FAIL on error. The original error is always re-raised. */
diff --git a/modules/core/src/main/scala/com/flowforge/core/impl/InMemoryPipelineBuilder.scala b/modules/core/src/main/scala/com/flowforge/core/impl/InMemoryPipelineBuilder.scala
index a784cc79..4f372a5a 100644
--- a/modules/core/src/main/scala/com/flowforge/core/impl/InMemoryPipelineBuilder.scala
+++ b/modules/core/src/main/scala/com/flowforge/core/impl/InMemoryPipelineBuilder.scala
@@ -1,9 +1,8 @@
-// scalafix:off DisableSyntax.isInstanceOf
 package com.flowforge.core.impl
 
 import cats.data.Kleisli
 import com.flowforge.core.algebra.{ DataAlgebra, DataDecoder, DataEncoder, EffectSystem }
-import com.flowforge.core.exec.{ ExecutableStage, StageComposer }
+import com.flowforge.core.exec.{ StageChain, StageComposer, StageKind }
 import com.flowforge.core.types._
 import com.flowforge.framework.{ Pipeline, PipelineMetadata }
 
@@ -27,7 +26,7 @@ class InMemoryPipelineBuilder[F[_]: EffectSystem] private (
     new InMemoryTypedBuilder[F, Unit, Unit](
       name = name,
       dataAlgebra = dataAlgebra,
-      stages = List.empty,
+      stages = StageChain.empty[F, Unit],
     )(EffectSystem[F])
 
   /**
@@ -40,7 +39,7 @@ class InMemoryPipelineBuilder[F[_]: EffectSystem] private (
     new InMemoryStreamBuilder[F, A, A](
       name = name,
       dataAlgebra = dataAlgebra,
-      stages = List.empty,
+      stages = StageChain.empty[F, fs2.Stream[F, A]],
     )(EffectSystem[F])
 }
 
@@ -55,7 +54,7 @@ class InMemoryPipelineBuilder[F[_]: EffectSystem] private (
 class InMemoryTypedBuilder[F[_], In, Out] private[impl] (
   private[impl] val name: String,
   private[impl] val dataAlgebra: InMemoryDataAlgebra[F],
-  private[impl] val stages: List[InMemoryStage[F, _, _]],
+  private[impl] val stages: StageChain[F, In, Out],
   private[impl] val description: String = "",
   private[impl] val config: Option[PipelineConfig] = None,
 )(implicit
@@ -72,11 +71,22 @@ class InMemoryTypedBuilder[F[_], In, Out] private[impl] (
    *
    * Every stage method used to call the constructor with three of the five fields, so `withDescription` and
    * `withConfig` were discarded by the next stage added. Carrying the fields in one place removes that.
+   *
+   * The arrow has to read what the builder currently produces, so the stage chain and the builder's own type
+   * parameters cannot drift apart.
    */
-  private[impl] def advance[In2, Out2](
-    stage: InMemoryStage[F, _, _],
-  ): InMemoryTypedBuilder[F, In2, Out2] =
-    new InMemoryTypedBuilder[F, In2, Out2](name, dataAlgebra, stages :+ stage, description, config)(ef)
+  private[impl] def advance[Out2](
+    stageName: String,
+    kind: StageKind,
+    arrow: Kleisli[F, Out, Out2],
+  ): InMemoryTypedBuilder[F, In, Out2] =
+    new InMemoryTypedBuilder[F, In, Out2](
+      name,
+      dataAlgebra,
+      stages.andThen(stageName, kind, arrow),
+      description,
+      config,
+    )(ef)
 
   /**
    * Add a streaming data source with fs2.Stream processing.
@@ -89,15 +99,15 @@ class InMemoryTypedBuilder[F[_], In, Out] private[impl] (
     source: DataSource,
     decoder: com.flowforge.core.algebra.DataDecoder[C],
   )(implicit atStart: Out =:= Unit,
-  ): InMemoryTypedBuilder[F, Unit, DataAlgebra.Dataset[C]] = {
-    val _ = atStart
+  ): InMemoryTypedBuilder[F, In, DataAlgebra.Dataset[C]] = {
     val stage = InMemoryStage.StreamingSource[F, C](
       name = s"stream-source-${stages.size}",
       description = s"Stream from ${source.format} with fs2",
       source = source,
       execute = Kleisli(_ => dataAlgebra.read(source)(decoder)),
     )
-    advance[Unit, DataAlgebra.Dataset[C]](stage)
+    // `=:=` is a function in both Scala versions, so the source's Unit input is adapted rather than cast.
+    advance[DataAlgebra.Dataset[C]](stage.name, StageKind.Source, stage.execute.local[Out](atStart))
   }
 
   /**
@@ -109,7 +119,7 @@ class InMemoryTypedBuilder[F[_], In, Out] private[impl] (
       description = "Memory-safe transformation with fs2",
       execute = Kleisli(transform),
     )
-    advance[In, C](stage)
+    advance[C](stage.name, StageKind.Transform, stage.execute)
   }
 
   /**
@@ -121,14 +131,14 @@ class InMemoryTypedBuilder[F[_], In, Out] private[impl] (
     // record type and the value type, and the stage types therefore did not line up.
     val kleisliPipeline = StageComposer.compose[F, In, Out](
       pipelineName = name,
-      stages = stages.map(st => ExecutableStage[F](st.name, st.asKleisli)),
+      stages = stages,
     )(ef)
 
     val metadata = PipelineMetadata(
       name = name,
-      stages = stages.map(_.name),
-      transformations = stages.count(_.isInstanceOf[InMemoryStage.Transform[F, _, _]]),
-      qualityChecks = stages.count(_.isInstanceOf[InMemoryStage.Quality[F, _]]),
+      stages = stages.names,
+      transformations = stages.count(StageKind.Transform),
+      qualityChecks = stages.count(StageKind.Quality),
       tags = Map(
         "engine"      -> "inmemory",
         "streaming"   -> "fs2",
@@ -166,7 +176,7 @@ object InMemoryTypedBuilder {
         description = "Batch transformation",
         execute = Kleisli(data => ef.pure(transform(data))),
       )
-      builder.advance[In, DataAlgebra.Dataset[C]](stage)
+      builder.advance[DataAlgebra.Dataset[C]](stage.name, StageKind.Transform, stage.execute)
     }
 
     /**
@@ -187,7 +197,7 @@ object InMemoryTypedBuilder {
         execute =
           Kleisli(data => ef.flatMap(algebra.validate(data, contract))(result => ef.pure(result.data))),
       )
-      builder.advance[In, DataAlgebra.Dataset[E]](stage)
+      builder.advance[DataAlgebra.Dataset[E]](stage.name, StageKind.Quality, stage.execute)
     }
 
     /**
@@ -206,7 +216,7 @@ object InMemoryTypedBuilder {
         sink = sink,
         execute = Kleisli(data => ef.flatMap(algebra.write(data, sink, options)(encoder))(_ => ef.pure(()))),
       )
-      builder.advance[In, Unit](stage)
+      builder.advance[Unit](stage.name, StageKind.Sink, stage.execute)
     }
   }
 }
@@ -222,7 +232,7 @@ object InMemoryTypedBuilder {
 class InMemoryStreamBuilder[F[_], In, Out] private[impl] (
   private val name: String,
   private val dataAlgebra: InMemoryDataAlgebra[F],
-  private val stages: List[InMemoryStage[F, _, _]],
+  private val stages: StageChain[F, fs2.Stream[F, In], fs2.Stream[F, Out]],
 )(implicit
   ef: EffectSystem[F]) {
 
@@ -238,7 +248,11 @@ class InMemoryStreamBuilder[F[_], In, Out] private[impl] (
         ef.pure(operation(stream))
       },
     )
-    new InMemoryStreamBuilder[F, In, B](name, dataAlgebra, stages :+ stage)(ef)
+    new InMemoryStreamBuilder[F, In, B](
+      name,
+      dataAlgebra,
+      stages.andThen(stageName, StageKind.Transform, stage.execute),
+    )(ef)
   }
 
   def buildStreaming(): Pipeline[F, fs2.Stream[F, In], fs2.Stream[F, Out]] = {
@@ -246,14 +260,14 @@ class InMemoryStreamBuilder[F[_], In, Out] private[impl] (
     // This used to return the input stream unchanged, ignoring every registered operation.
     val kleisliPipeline = StageComposer.compose[F, fs2.Stream[F, In], fs2.Stream[F, Out]](
       pipelineName = name,
-      stages = stages.map(st => ExecutableStage[F](st.name, st.asKleisli)),
+      stages = stages,
     )(ef)
 
     val metadata = PipelineMetadata(
       name = name,
-      stages = stages.map(_.name),
-      transformations = stages.size,
-      qualityChecks = 0,
+      stages = stages.names,
+      transformations = stages.count(StageKind.Transform),
+      qualityChecks = stages.count(StageKind.Quality),
       tags = Map(
         "engine"         -> "inmemory",
         "streaming"      -> "fs2",
diff --git a/modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkPipelineBuilder.scala b/modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkPipelineBuilder.scala
index 6f5383ae..2094b2fd 100644
--- a/modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkPipelineBuilder.scala
+++ b/modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkPipelineBuilder.scala
@@ -5,7 +5,7 @@ import cats.effect.Resource
 import cats.implicits._
 import com.flowforge.core.algebra.DataAlgebra.WriteOptions
 import com.flowforge.core.algebra.{ DataAlgebra, DataDecoder, EffectSystem }
-import com.flowforge.core.exec.{ ExecutableStage, StageComposer }
+import com.flowforge.core.exec.{ StageChain, StageComposer, StageKind }
 import com.flowforge.core.types._
 import com.flowforge.framework.{ Pipeline, PipelineMetadata }
 import org.apache.spark.sql.SparkSession
@@ -42,18 +42,18 @@ class SparkPipelineBuilder[F[_]: EffectSystem] private (
       name = name,
       sparkSession = sparkSession,
       dataAlgebra = dataAlgebra,
-      stages = List.empty,
+      stages = StageChain.empty[F, Unit],
     )
 
   /**
    * Create a runtime pipeline builder with Spark backend
    */
-  def runtime(name: String): SparkRuntimeBuilder[F] =
-    new SparkRuntimeBuilder[F](
+  def runtime(name: String): SparkRuntimeBuilder[F, Unit, Unit] =
+    new SparkRuntimeBuilder[F, Unit, Unit](
       name = name,
       sparkSession = sparkSession,
       dataAlgebra = dataAlgebra,
-      stages = List.empty,
+      stages = StageChain.empty[F, Unit],
     )
 }
 
@@ -61,12 +61,12 @@ class SparkPipelineBuilder[F[_]: EffectSystem] private (
  * Type-safe Spark pipeline builder with compile-time guarantees
  */
 class SparkTypedBuilder[F[_]: EffectSystem, In, Out] private[spark] (
-  private val name: String,
-  private val sparkSession: SparkSession,
-  private val dataAlgebra: DataAlgebra[F],
-  private val stages: List[SparkStage[F, _, _]],
-  private val description: String = "",
-  private val config: Option[PipelineConfig] = None) {
+  private[spark] val name: String,
+  private[spark] val sparkSession: SparkSession,
+  private[spark] val dataAlgebra: DataAlgebra[F],
+  private[spark] val stages: StageChain[F, In, Out],
+  private[spark] val description: String = "",
+  private[spark] val config: Option[PipelineConfig] = None) {
 
   def withDescription(desc: String): SparkTypedBuilder[F, In, Out] =
     new SparkTypedBuilder(name, sparkSession, dataAlgebra, stages, desc, config)
@@ -74,20 +74,39 @@ class SparkTypedBuilder[F[_]: EffectSystem, In, Out] private[spark] (
   def withConfig(cfg: PipelineConfig): SparkTypedBuilder[F, In, Out] =
     new SparkTypedBuilder(name, sparkSession, dataAlgebra, stages, description, Some(cfg))
 
+  /** Append a stage. Its arrow has to read what the builder currently produces. */
+  private[spark] def advance[Out2](
+    stageName: String,
+    kind: StageKind,
+    arrow: Kleisli[F, Out, Out2],
+  ): SparkTypedBuilder[F, In, Out2] =
+    new SparkTypedBuilder[F, In, Out2](
+      name,
+      sparkSession,
+      dataAlgebra,
+      stages.andThen(stageName, kind, arrow),
+      description,
+      config,
+    )
+
   /**
-   * Add a Spark-optimized data source
+   * Add a Spark-optimized data source.
+   *
+   * A source takes no input, so the `Out =:= Unit` evidence restricts it to a builder that has not produced a
+   * value yet. At `typed(name)` the builder's `Out` is already `Unit`, so no caller names the evidence.
    */
   def addSource[C](
     source: DataSource,
     decoder: com.flowforge.core.algebra.DataDecoder[C],
-  ): SparkTypedBuilder[F, Unit, C] = {
+  )(implicit atStart: Out =:= Unit,
+  ): SparkTypedBuilder[F, In, DataAlgebra.Dataset[C]] = {
     val stage = SparkStage.Source[F, C](
       name = s"spark-source-${stages.size}",
       description = s"Read from ${source.format} using Spark",
       source = source,
       execute = Kleisli(_ => dataAlgebra.read(source)(decoder)),
     )
-    new SparkTypedBuilder[F, Unit, C](name, sparkSession, dataAlgebra, stages :+ stage)
+    advance[DataAlgebra.Dataset[C]](stage.name, StageKind.Source, stage.execute.local[Out](atStart))
   }
 
   /**
@@ -99,70 +118,26 @@ class SparkTypedBuilder[F[_]: EffectSystem, In, Out] private[spark] (
       description = "Spark distributed transformation",
       execute = Kleisli(transform),
     )
-    new SparkTypedBuilder[F, In, C](name, sparkSession, dataAlgebra, stages :+ stage)
-  }
-
-  /**
-   * Add a data quality check using Spark's distributed validation.
-   *
-   * The decoder is what lets the engine apply the contract to every record. See `DataAlgebra.validate`.
-   */
-  def addQualityCheck(
-    contract: com.flowforge.core.types.PipelineTypes.DataContract[Out],
-  )(implicit decoder: DataDecoder[Out],
-  ): SparkTypedBuilder[F, In, Out] = {
-    EffectSystem[F]
-    val stage = SparkStage.Quality[F, Out](
-      name = s"spark-quality-${stages.size}",
-      description = "Spark distributed quality validation",
-      contract = contract,
-      execute = Kleisli { data =>
-        for {
-          res <- dataAlgebra.validate(data, contract)
-        } yield res.data
-      },
-    )
-    new SparkTypedBuilder[F, In, Out](name, sparkSession, dataAlgebra, stages :+ stage)
-  }
-
-  /**
-   * Add a sink that uses Spark's distributed writing capabilities
-   */
-  def addSink(
-    sink: DataSink,
-    encoder: com.flowforge.core.algebra.DataEncoder[Out],
-    options: WriteOptions = WriteOptions.default,
-  ): SparkTypedBuilder[F, In, Unit] = {
-    val stage = SparkStage.Sink[F, Out](
-      name = s"spark-sink-${stages.size}",
-      description = s"Write to ${sink.format} using Spark",
-      sink = sink,
-      execute = Kleisli { data =>
-        for {
-          _ <- dataAlgebra.write(data, sink, options)(encoder)
-        } yield ()
-      },
-    )
-    new SparkTypedBuilder[F, In, Unit](name, sparkSession, dataAlgebra, stages :+ stage)
+    advance[C](stage.name, StageKind.Transform, stage.execute)
   }
 
   /**
    * Build the final pipeline with Spark optimizations
    */
   def build(): Pipeline[F, In, Out] = {
-    // Composition belongs to StageComposer, which is the one place that erases a stage arrow. This builder
-    // used to do its own fold and its own two casts, so the erasure existed in two places and only one of
-    // them was tested.
+    // Composition belongs to StageComposer, which is the one place that answers how a chain of stages runs.
+    // This builder used to do its own fold and its own two casts, so the erasure existed in two places and
+    // only one of them was tested.
     val typedPipeline = StageComposer.compose[F, In, Out](
       pipelineName = name,
-      stages = stages.map(st => ExecutableStage[F](st.name, st.asKleisli)),
+      stages = stages,
     )
 
     val metadata = PipelineMetadata(
       name = name,
-      stages = stages.map(_.name),
-      transformations = stages.collect { case _: SparkStage.Transform[_, _, _] => 1 }.size,
-      qualityChecks = stages.collect { case _: SparkStage.Quality[_, _] => 1 }.size,
+      stages = stages.names,
+      transformations = stages.count(StageKind.Transform),
+      qualityChecks = stages.count(StageKind.Quality),
       tags = Map(
         "engine"        -> "spark",
         "type_safe"     -> "true",
@@ -175,38 +150,97 @@ class SparkTypedBuilder[F[_]: EffectSystem, In, Out] private[spark] (
 }
 
 /**
- * Runtime pipeline builder for dynamic pipeline construction
+ * Stages that only make sense once the builder is carrying a dataset.
+ *
+ * A quality check validates records and a sink writes records, so both need the pipeline to be holding a
+ * `Dataset`. Keeping them here rather than on the class is what lets them say so: the implicit class only
+ * resolves for a builder whose `Out` is a dataset. On the class they were declared over the builder's `Out`
+ * while their stage arrows were over `Dataset[Out]`, and the composer's cast hid the mismatch.
  */
-class SparkRuntimeBuilder[F[_]: EffectSystem] private[spark] (
+object SparkTypedBuilder {
+
+  implicit class DatasetOps[F[_]: EffectSystem, In, E](
+    builder: SparkTypedBuilder[F, In, DataAlgebra.Dataset[E]]) {
+
+    /**
+     * Add a data quality check using Spark's distributed validation.
+     *
+     * The decoder is what lets the engine apply the contract to every record. See `DataAlgebra.validate`.
+     */
+    def addQualityCheck(
+      contract: com.flowforge.core.types.PipelineTypes.DataContract[E],
+    )(implicit decoder: DataDecoder[E],
+    ): SparkTypedBuilder[F, In, DataAlgebra.Dataset[E]] = {
+      val _ = decoder
+      val stage = SparkStage.Quality[F, E](
+        name = s"spark-quality-${builder.stages.size}",
+        description = "Spark distributed quality validation",
+        contract = contract,
+        execute = Kleisli(data => builder.dataAlgebra.validate(data, contract).map(_.data)),
+      )
+      builder.advance[DataAlgebra.Dataset[E]](stage.name, StageKind.Quality, stage.execute)
+    }
+
+    /**
+     * Add a sink that uses Spark's distributed writing capabilities
+     */
+    def addSink(
+      sink: DataSink,
+      encoder: com.flowforge.core.algebra.DataEncoder[E],
+      options: WriteOptions = WriteOptions.default,
+    ): SparkTypedBuilder[F, In, Unit] = {
+      val stage = SparkStage.Sink[F, E](
+        name = s"spark-sink-${builder.stages.size}",
+        description = s"Write to ${sink.format} using Spark",
+        sink = sink,
+        execute = Kleisli(data => builder.dataAlgebra.write(data, sink, options)(encoder).void),
+      )
+      builder.advance[Unit](stage.name, StageKind.Sink, stage.execute)
+    }
+  }
+}
+
+/**
+ * Runtime pipeline builder for dynamic pipeline construction.
+ *
+ * "Runtime" names where the stage operations come from, not whether the stage types line up. The chain still
+ * carries the type between each pair of stages, so `buildRuntime` returns the pipeline's own `In` and `Out`
+ * rather than the `Any` the old cast produced.
+ */
+class SparkRuntimeBuilder[F[_]: EffectSystem, In, Out] private[spark] (
   private val name: String,
   private val sparkSession: SparkSession,
   private val dataAlgebra: DataAlgebra[F],
-  private val stages: List[SparkStage[F, _, _]]) {
+  private val stages: StageChain[F, In, Out]) {
 
-  def addDynamicStage[A, B](
+  def addDynamicStage[B](
     stageName: String,
-    operation: DataAlgebra.Dataset[A] => F[DataAlgebra.Dataset[B]],
-  ): SparkRuntimeBuilder[F] = {
-    val stage = SparkStage.Dynamic[F, DataAlgebra.Dataset[A], DataAlgebra.Dataset[B]](
+    operation: Out => F[B],
+  ): SparkRuntimeBuilder[F, In, B] = {
+    val stage = SparkStage.Dynamic[F, Out, B](
       name = stageName,
       description = "Dynamic Spark operation",
       execute = Kleisli(operation),
     )
-    new SparkRuntimeBuilder[F](name, sparkSession, dataAlgebra, stages :+ stage)
+    new SparkRuntimeBuilder[F, In, B](
+      name,
+      sparkSession,
+      dataAlgebra,
+      stages.andThen(stageName, StageKind.Transform, stage.execute),
+    )
   }
 
-  def buildRuntime(): Pipeline[F, Any, Any] = {
-    // Runtime composition - less type safety but more flexibility
-    val kleisliPipeline = StageComposer.compose[F, Any, Any](
+  def buildRuntime(): Pipeline[F, In, Out] = {
+    val kleisliPipeline = StageComposer.compose[F, In, Out](
       pipelineName = name,
-      stages = stages.map(st => ExecutableStage[F](st.name, st.asKleisli)),
+      stages = stages,
     )
 
     val metadata = PipelineMetadata(
       name = name,
-      stages = stages.map(_.name),
-      transformations = stages.size,
-      qualityChecks = 0,
+      stages = stages.names,
+      transformations = stages.count(StageKind.Transform),
+      qualityChecks = stages.count(StageKind.Quality),
       tags = Map(
         "engine"    -> "spark",
         "type_safe" -> "false",

From 5e38625f22f2c0213ff4d9b6752cef205b33d93b Mon Sep 17 00:00:00 2001
From: Vitthal Mirji <vitthalmirji@gmail.com>
Date: Tue, 6 Oct 2026 17:32:01 +0200
Subject: [PATCH 3/4] perf(core): share the stage metadata list instead of
 copying it per append

---
 .../com/flowforge/core/exec/StageChain.scala  | 45 +++++++++++++------
 .../flowforge/core/exec/StageChainSpec.scala  | 10 +++++
 2 files changed, 42 insertions(+), 13 deletions(-)

diff --git a/modules/core/src/main/scala/com/flowforge/core/exec/StageChain.scala b/modules/core/src/main/scala/com/flowforge/core/exec/StageChain.scala
index c53d9c2f..3671ccec 100644
--- a/modules/core/src/main/scala/com/flowforge/core/exec/StageChain.scala
+++ b/modules/core/src/main/scala/com/flowforge/core/exec/StageChain.scala
@@ -38,8 +38,8 @@ object StageKind {
  * 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.
+ * 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
@@ -47,17 +47,24 @@ object StageKind {
  */
 sealed abstract class StageChain[F[_], In, Out] extends Product with Serializable {
 
-  /** Each stage's name and kind, in the order they run. */
-  private[exec] def entries: List[StageChain.Entry]
+  /**
+   * 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]
 
-  /** The stage names, in the order they run. */
-  final def names: List[String] = entries.map(_.name)
+  /** How many stages the chain holds. Held per link, because a builder reads it on every append. */
+  def size: Int
 
-  /** How many stages the chain holds. */
-  final def size: Int = entries.size
+  /** 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 = entries.count(_.kind == kind)
+  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](
@@ -72,6 +79,13 @@ sealed abstract class StageChain[F[_], In, Out] extends Product with Serializabl
    *
    * `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]].
+   *
+   * Building the arrow descends one frame per stage, so the chain's depth is bounded by the JVM stack: 10k
+   * stages build and run, 50k overflow. A left fold would not have that bound, but folding needs the chain
+   * reversed, and reversing a type-aligned sequence needs the same cast this type exists to remove. The bound
+   * is the one being accepted, and 10k is far past a pipeline anyone would run: every stage carries a span
+   * and a START/COMPLETE pair, so a chain near the bound is unusable for reasons that have nothing to do with
+   * the stack. [[StageChainSpec]] pins 10k so the bound cannot quietly drop.
    */
   private[exec] def arrow(decorate: StageDecorator[F])(implicit F: EffectSystem[F]): Kleisli[F, In, Out]
 }
@@ -93,7 +107,9 @@ object StageChain {
    */
   private final case class Identity[F[_], A]() extends StageChain[F, A, A] {
 
-    private[exec] def entries: List[Entry] = Nil
+    private[exec] def entriesReversed: List[Entry] = Nil
+
+    val size: Int = 0
 
     private[exec] def arrow(
       decorate: StageDecorator[F],
@@ -107,9 +123,12 @@ object StageChain {
     stage: Kleisli[F, Mid, Out])
       extends StageChain[F, In, Out] {
 
-    // Computed once per append rather than per read: `names`, `size` and `count` are all called while a
-    // pipeline is being built, and the next stage's name is derived from `size`.
-    private[exec] val entries: List[Entry] = init.entries :+ entry
+    // 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],
diff --git a/modules/core/src/test/scala/com/flowforge/core/exec/StageChainSpec.scala b/modules/core/src/test/scala/com/flowforge/core/exec/StageChainSpec.scala
index 047af58b..c8641ddd 100644
--- a/modules/core/src/test/scala/com/flowforge/core/exec/StageChainSpec.scala
+++ b/modules/core/src/test/scala/com/flowforge/core/exec/StageChainSpec.scala
@@ -47,6 +47,16 @@ class StageChainSpec extends AnyFunSuite with Matchers {
     withSink.size shouldBe 3
   }
 
+  test("a chain long enough to be generated rather than written out still builds and runs") {
+    // 10k is far past any hand-written pipeline and well past what a generator would produce. It is here
+    // because `arrow` descends one frame per link, so this is the bound that assertion rests on.
+    val deep = (1 to 10000).foldLeft(StageChain.empty[IO, Int]) { (chain, i) =>
+      chain.andThen(s"add-$i", StageKind.Transform, Kleisli((n: Int) => IO.pure(n + 1)))
+    }
+    deep.size shouldBe 10000
+    deep.arrow(plain).run(0).unsafeRunSync() shouldBe 10000
+  }
+
   test("a stage that does not read the previous stage's output is rejected") {
     assertDoesNotCompile(
       """

From 07e1242a5c214bc515b7b95dcbc26e0a92937d63 Mon Sep 17 00:00:00 2001
From: Vitthal Mirji <vitthalmirji@gmail.com>
Date: Tue, 6 Oct 2026 17:51:12 +0200
Subject: [PATCH 4/4] fix: keep deep stage chains off the JVM stack

---
 .../com/flowforge/core/exec/StageChain.scala  | 22 ++++++----
 .../flowforge/core/exec/StageChainSpec.scala  | 40 ++++++++++++++++---
 2 files changed, 49 insertions(+), 13 deletions(-)

diff --git a/modules/core/src/main/scala/com/flowforge/core/exec/StageChain.scala b/modules/core/src/main/scala/com/flowforge/core/exec/StageChain.scala
index 3671ccec..72308c97 100644
--- a/modules/core/src/main/scala/com/flowforge/core/exec/StageChain.scala
+++ b/modules/core/src/main/scala/com/flowforge/core/exec/StageChain.scala
@@ -80,12 +80,14 @@ sealed abstract class StageChain[F[_], In, Out] extends Product with Serializabl
    * `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]].
    *
-   * Building the arrow descends one frame per stage, so the chain's depth is bounded by the JVM stack: 10k
-   * stages build and run, 50k overflow. A left fold would not have that bound, but folding needs the chain
-   * reversed, and reversing a type-aligned sequence needs the same cast this type exists to remove. The bound
-   * is the one being accepted, and 10k is far past a pipeline anyone would run: every stage carries a span
-   * and a START/COMPLETE pair, so a chain near the bound is unusable for reasons that have nothing to do with
-   * the stack. [[StageChainSpec]] pins 10k so the bound cannot quietly drop.
+   * 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]
 }
@@ -133,7 +135,13 @@ object StageChain {
     private[exec] def arrow(
       decorate: StageDecorator[F],
     )(implicit F: EffectSystem[F],
-    ): Kleisli[F, In, Out] = init.arrow(decorate).andThen(decorate(entry.name, stage))
+    ): 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))
+    }
   }
 }
 
diff --git a/modules/core/src/test/scala/com/flowforge/core/exec/StageChainSpec.scala b/modules/core/src/test/scala/com/flowforge/core/exec/StageChainSpec.scala
index c8641ddd..f247bce7 100644
--- a/modules/core/src/test/scala/com/flowforge/core/exec/StageChainSpec.scala
+++ b/modules/core/src/test/scala/com/flowforge/core/exec/StageChainSpec.scala
@@ -21,6 +21,32 @@ class StageChainSpec extends AnyFunSuite with Matchers {
     def apply[A, B](stageName: String, stage: Kleisli[IO, A, B]): Kleisli[IO, A, B] = stage
   }
 
+  /**
+   * Runs `body` on a thread with a 512k stack and hands back whatever it produced.
+   *
+   * A deep-chain test on the default stack measures the runner, not the code: the JVM's main thread gets 8MB
+   * on macOS and about 1MB on the Linux and Windows runners. Naming the stack size makes the bound the test
+   * asserts the same everywhere. `StackOverflowError` is not `NonFatal`, so `Try` would not catch it.
+   */
+  private def onSmallStack[A](body: => A): A = {
+    var outcome: Either[Throwable, A] = Left(new IllegalStateException("thread never ran"))
+    val thread = new Thread(
+      null,
+      () =>
+        outcome =
+          try Right(body)
+          catch { case e: Throwable => Left(e) },
+      "small-stack",
+      512L * 1024L,
+    )
+    thread.start()
+    thread.join()
+    outcome match {
+      case Right(value) => value
+      case Left(e)      => throw e
+    }
+  }
+
   private def chain: StageChain[IO, Int, String] =
     StageChain
       .empty[IO, Int]
@@ -47,14 +73,16 @@ class StageChainSpec extends AnyFunSuite with Matchers {
     withSink.size shouldBe 3
   }
 
-  test("a chain long enough to be generated rather than written out still builds and runs") {
-    // 10k is far past any hand-written pipeline and well past what a generator would produce. It is here
-    // because `arrow` descends one frame per link, so this is the bound that assertion rests on.
-    val deep = (1 to 10000).foldLeft(StageChain.empty[IO, Int]) { (chain, i) =>
+  test("a chain deeper than the thread's stack still builds and runs") {
+    // Runs on a small stack on purpose. An arrow that walked its links eagerly overflowed here between 2k
+    // and 4k, and the depth it survived depended on the thread rather than the chain: the same 10k chain
+    // passed on macOS and overflowed on Linux and Windows CI. 50k on 512k of stack leaves no room for that
+    // to pass by accident.
+    val deep = (1 to 50000).foldLeft(StageChain.empty[IO, Int]) { (chain, i) =>
       chain.andThen(s"add-$i", StageKind.Transform, Kleisli((n: Int) => IO.pure(n + 1)))
     }
-    deep.size shouldBe 10000
-    deep.arrow(plain).run(0).unsafeRunSync() shouldBe 10000
+    deep.size shouldBe 50000
+    onSmallStack(deep.arrow(plain).run(0).unsafeRunSync()) shouldBe 50000
   }
 
   test("a stage that does not read the previous stage's output is rejected") {