Scan

Scan folds its own inputs into a running accumulator across pipeline runs, via the binary operation stage op — the running-total counterpart of Reduce and Fold. Unlike that cycles family, Scan has no inner stage and no cycle: it consumes one external input per outer run and threads its own accumulator from one run to the next.

The first input becomes the accumulator unchanged, without invoking op; each subsequent input is folded via op((accumulator, input)), and the result — the new running total — is emitted immediately.

import h8io.stages.*
import h8io.stages.base.*
import h8io.stages.std.*

object Sum extends Fn[(Int, Int), Int] {
  override protected def f(in: (Int, Int)): Int = in._1 + in._2
}

val running = Scan(Sum)
// running: Scan[Int, Nothing] = Scan(op = <function1>)
val y1 = running(1)
// y1: Yield[Int, Int, Nothing] = Some(
//   out = 1,
//   status = Success,
//   evolution = Evolution(value = 1, opEvolution = <function1>)
// )
val y2 = y1.evolve()(2)
// y2: Yield[Int, Int, Nothing] = Some(
//   out = 3,
//   status = Success,
//   evolution = Evolution(value = 3, opEvolution = <function1>)
// )
val y3 = y2.evolve()(3)
// y3: Yield[Int, Int, Nothing] = Some(
//   out = 6,
//   status = Success,
//   evolution = Evolution(value = 6, opEvolution = <function1>)
// )

1 seeds the accumulator; 2 and 3 are folded in, so each run reports the running total: 1, 3, 6.

When op is applied but itself yields no output — for example, it filters some inputs out — the accumulator is left unchanged rather than discarded, exactly as in Reduce/Fold, and Scan itself yields nothing for that run:

object SumOnlyEven extends StaticStage[(Int, Int), Int, Nothing] {
  override protected def process(in: (Int, Int)): StaticYield[Int, Nothing] = {
    val (acc, out) = in
    if (out % 2 == 0) StaticYield.Some(acc + out, Status.Success)
    else              StaticYield.None(Status.Success)
  }
}

val filtered = Scan(SumOnlyEven)
// filtered: Scan[Int, Nothing] = Scan(op = <function1>)
val z1 = filtered(10)
// z1: Yield[Int, Int, Nothing] = Some(
//   out = 10,
//   status = Success,
//   evolution = Evolution(value = 10, opEvolution = <function1>)
// )
val z2 = z1.evolve()(3)
// z2: Yield[Int, Int, Nothing] = None(
//   status = Success,
//   evolution = Evolution(value = 10, opEvolution = <function1>)
// )
val z3 = z2.evolve()(4)
// z3: Yield[Int, Int, Nothing] = Some(
//   out = 14,
//   status = Success,
//   evolution = Evolution(value = 14, opEvolution = <function1>)
// )

10 seeds the accumulator. 3 is odd, so SumOnlyEven declines: z2 is None and the accumulator stays 10, untouched by the declined fold. 4 is even, so it folds against the still-10 accumulator: z3 is Some(14, ...).

Whenever a run's status is Status.Complete, the next generation resets to a fresh, unseeded Scan instead of continuing to accumulate — mirroring how Countdown resets on any non-Success status. Complete marks one logical unit of work as finished, so the next input seeds a new one from scratch rather than folding into whatever was accumulated before:

final class SumUntil(limit: Int) extends Stage[(Int, Int), Int, Nothing] with Evolution[(Int, Int), Int, Nothing] {
  override def apply(in: (Int, Int)): Yield[(Int, Int), Int, Nothing] = {
    val (acc, out) = in
    val sum = acc + out
    val status: Status[Nothing] = if (sum >= limit) Status.complete else Status.Success
    Yield.Some(sum, status, this)
  }
  override def skip(): Evolution[(Int, Int), Int, Nothing] = this
  override def evolve(status: Status[?]): Stage[(Int, Int), Int, Nothing] = this
  override def dispose(): Unit = ()
}

val bounded = Scan(new SumUntil(10))
// bounded: Scan[Int, Nothing] = Scan(op = <function1>)
val w1 = bounded(4)
// w1: Yield[Int, Int, Nothing] = Some(
//   out = 4,
//   status = Success,
//   evolution = Evolution(value = 4, opEvolution = <function1>)
// )
val w2 = w1.evolve()(3)
// w2: Yield[Int, Int, Nothing] = Some(
//   out = 7,
//   status = Success,
//   evolution = Evolution(value = 7, opEvolution = <function1>)
// )
val w3 = w2.evolve()(9) // 4 + 3 = 7, then 7 + 9 = 16 >= 10, so Complete
// w3: Yield[Int, Int, Nothing] = Some(
//   out = 16,
//   status = Complete(),
//   evolution = Evolution(value = 16, opEvolution = <function1>)
// )
val w4 = w3.evolve()(2) // resets: seeds fresh with 2 instead of folding 16 + 2
// w4: Yield[Int, Int, Nothing] = Some(
//   out = 2,
//   status = Success,
//   evolution = Evolution(value = 2, opEvolution = <function1>)
// )

w3 reaches 16 and signals Complete, so w4 does not fold 2 into it: it seeds a brand new accumulator, Some(2, Success, ...).