Stage

Stage is the fundamental processing unit of a pipeline.
It is a function I => Yield[I, O, E] that transforms a single input value into a Yield — an optional output, a Status, and an Evolution that decides what the pipeline looks like for the next run.

Stages are contravariant in their input type I and covariant in their output type O and error type E. This variance enables safe composition: a stage that accepts a wider input type can be used wherever a more specific one is expected, and outputs can flow naturally from one stage to the next.

import h8io.stages.*

The Lifecycle: apply and skip

Every Stage takes exactly one of three paths during a pipeline run.

apply — the active path. The stage receives a real input value and returns a Yield. It is free to perform side effects, update internal state, and produce any output it likes.

skip — the bypassed path. Any stage that participates in a pipeline run but does not process the current input must take this path: it returns the Evolution it would have returned had it run, without consuming any value. Like apply, it may perform side effects — a decorator, for example, may advance the lifecycle of the inner stage it owns. Common triggers are an upstream stage producing Yield.None (nothing to pass downstream) or a non-inclusive binary operation that excludes this branch.

skip exists precisely so that stages further downstream still get a chance to evolve correctly even when they are not directly executed. The Diagram explains this with a concrete example involving Stage 3-1, which is skipped in the first generation but still needs to evolve into Stage 3-2.

apply that throws — the failure path. No Yield — and therefore no Evolution — is produced, so nobody can release the stage's resources from the outside: a stage that throws from apply must release its own resources before the exception escapes, and is permanently unusable afterwards. The core takes no other part in exception handling; for the simplest — stateless — stages the lib module offers the Safe crutch.

Here is a minimal stage that doubles its input on the active path and supplies its evolution on the skipped path:

object Double extends Stage[Int, Int, Nothing] {
  private def evo: Evolution[Int, Int, Nothing] = new Evolution[Int, Int, Nothing] {
    override def evolve(status: Status[?]): Stage[Int, Int, Nothing] = Double
    override def dispose(): Unit = ()
  }

  override def apply(in: Int): Yield[Int, Int, Nothing] =
    Yield.Some(in * 2, Status.Success, evo)

  override def skip(): Evolution[Int, Int, Nothing] = evo
}

Double(7)
// res0: Yield[Int, Int, Nothing] = Some(
//   out = 14,
//   status = Success,
//   evolution = repl.MdocSession$MdocApp$Double$$anon$1@49fc943d
// )
Double.skip()
// res1: Evolution[Int, Int, Nothing] = repl.MdocSession$MdocApp$Double$$anon$1@7ccade74

Building Pipelines with ~>

~> composes two stages: the output of the left stage becomes the input of the right stage. The composed node itself is internal to the core — what you get back is a Stage, so further stages can be appended with additional ~> calls:

object ToString extends Stage[Int, String, Nothing] {
  private def evo: Evolution[Int, String, Nothing] = new Evolution[Int, String, Nothing] {
    override def evolve(status: Status[?]): Stage[Int, String, Nothing] = ToString
    override def dispose(): Unit = ()
  }

  override def apply(in: Int): Yield[Int, String, Nothing] =
    Yield.Some(in.toString, Status.Success, evo)

  override def skip(): Evolution[Int, String, Nothing] = evo
}

object Shout extends Stage[String, String, Nothing] {
  private def evo: Evolution[String, String, Nothing] = new Evolution[String, String, Nothing] {
    override def evolve(status: Status[?]): Stage[String, String, Nothing] = Shout
    override def dispose(): Unit = ()
  }

  override def apply(in: String): Yield[String, String, Nothing] =
    Yield.Some(in.toUpperCase + "!", Status.Success, evo)

  override def skip(): Evolution[String, String, Nothing] = evo
}

val pipeline = Double ~> ToString ~> Shout
// pipeline: Stage[Int, String, Nothing] = AndThen(
//   upstream = AndThen(upstream = <function1>, downstream = <function1>),
//   downstream = <function1>
// )

The static type of pipeline is Stage[Int, String, Nothing]: it accepts the input of the first stage and produces the output of the last one. The intermediate types — Int between the first two stages, String between the last two — do not appear in the type of the pipeline at all.

Running the pipeline produces a single Yield that combines the status and evolution of all three stages:

pipeline(5)
// res2: Yield[Int, String, Nothing] = Some(
//   out = "10!",
//   status = Success,
//   evolution = AndThen(
//     upstream = repl.MdocSession$MdocApp$Shout$$anon$3@14f4219,
//     downstream = AndThen(
//       upstream = repl.MdocSession$MdocApp$ToString$$anon$2@1f27c675,
//       downstream = repl.MdocSession$MdocApp$Double$$anon$1@2d9db618
//     )
//   )
// )

When the upstream stage produces Yield.None, the downstream stage is not applied; instead it is wired into the combined evolution so that it will be called when the next input arrives.

Terminal Execution

The core model ends at Yield: whoever terminates a pipeline is responsible for disposing the evolution of the final Yield. The reference terminal driver is the execute extension method from h8io.stages.base — the one-shot path that applies the stage to an input, immediately disposes the Evolution, and returns a plain Outcome with no continuation:

import h8io.stages.base.*

val outcome = pipeline.execute(5)
// outcome: Outcome[String, Nothing] = Some(
//   out = "10!",
//   status = Success,
//   disposeFailure = None
// )

See Outcome for the details, including how disposal failures are reported.