///|
/// Run thunks concurrently, each already carrying its failure policy in
/// its type: a thunk resolves to `Result[T, WorkflowError]` (wrap a call
/// in `attempt`), so one agent's failure lands in its own slot without
/// disturbing siblings — tokens already spent on the others stay spent.
/// Anything a thunk RAISES is deliberately not caught: engine bugs and
/// cancellation fail the task group and cancel every peer, never demoted
/// to a slot a policy could quietly discard. Pair with `all_ok` /
/// `collect_ok` / `quorum` to choose the policy in one identifier.
pub async fn[T] parallel(
  thunks : Array[async () -> Result[T, WorkflowError]],
) -> Array[Result[T, WorkflowError]] {
  @async.all(thunks)
}

///|
/// Fail-fast fan-out: the first thunk to RAISE (an `agent` call
/// propagating its `AgentFailed`, an engine bug, cancellation) cancels
/// every sibling still in flight and propagates. The right shape when
/// later work is worthless without ALL of this stage — otherwise prefer
/// `parallel`, which keeps what succeeded.
pub async fn[T] parallel_all(thunks : Array[async () -> T]) -> Array[T] {
  @async.all(thunks)
}

///|
/// Fan items out through one async stage, one `Result` slot per item.
/// Multi-stage pipelines are function composition inside `run` — stages
/// need no barrier between them, so composing them per-item IS the
/// pipeline.
pub async fn[A, B] fan_out(
  items : Array[A],
  run : async (A) -> Result[B, WorkflowError],
) -> Array[Result[B, WorkflowError]] {
  parallel(items.map(item => () => run(item)))
}

///|
/// Fold the typed error channel into the return value: a workflow-level
/// failure lands in `Err`, while cancellation and engine bugs still
/// propagate — the distinction the whole failure model rests on.
///
/// It folds ANY raising step: `agent`, `agent_call` carrying an engine's
/// own input shape, an `agent_as` whose decode may reject, a `retry`
/// around any of them, or a whole multi-stage per-item pipeline. ONE free
/// function rather than a `try_` twin per entry point — the shape a
/// caller wants is a combinator, never another method.
///
/// The right shape at fan-out call sites, where one lost agent must not
/// poison its siblings.
pub async fn[T] attempt(step : async () -> T) -> Result[T, WorkflowError] {
  Ok(step()) catch {
    AgentFailed(..) as failure => Err(failure)
    CallBudgetExhausted(..) as failure => Err(failure)
    QuorumNotReached(..) as failure => Err(failure)
    error => raise error
  }
}

///|
/// Re-run one agent step while it keeps failing retriably: `max_retry`
/// extra attempts at most, spaced by `backoff`, raising the LAST attempt's
/// error when the budget runs out. Wraps a RAISING step (`agent`,
/// `agent_as`, or a whole per-item pipeline). Wrap it in `attempt`, not
/// the other way round: a step already folded into a `Result` raises
/// nothing there is left to retry on.
///
/// What counts as retriable is `worth_retrying` unless `retriable` says
/// otherwise; cancellation and engine bugs are never retried, they
/// propagate on the first raise.
///
/// `backoff` is the async runtime's own `RetryMethod`, and NAMING one
/// (`FixedDelay(500)`) needs `moonbitlang/async` in the caller's
/// `moon.pkg` — a re-export cannot lend it, since an aliased type still
/// resolves through the package that owns it. The default needs nothing.
///
/// Every attempt is a full LAUNCH: it queues for a slot, debits the launch
/// allowance, and charges its tokens — so `max_calls` must be sized with
/// re-attempts in mind, exactly as it must for resume. Journal replay is
/// unaffected: a failed attempt is recorded and never replayed, so a
/// resumed run replays the attempt that finally succeeded.
pub async fn[T] retry(
  attempt : async () -> T,
  max_retry? : Int = 1,
  backoff? : @async.RetryMethod = Immediate,
  retriable? : (WorkflowError) -> Bool = worth_retrying,
) -> T {
  @async.retry(
    backoff,
    max_retry~,
    // The runtime's own loop already treats cancellation as fatal; this
    // adds the workflow's vocabulary — anything that is not a typed
    // workflow failure is an engine bug and must not be re-run.
    fatal_error=error => {
      match error {
        AgentFailed(..) as failure => !retriable(failure)
        CallBudgetExhausted(..) as failure => !retriable(failure)
        QuorumNotReached(..) as failure => !retriable(failure)
        _ => true
      }
    },
    attempt,
  )
}

///|
/// Policy: every slot must have succeeded; re-raise the first failure
/// otherwise. Post-hoc fail-fast — siblings have already run to
/// completion; use `parallel_all` when failure should cancel them instead.
pub fn[T] all_ok(
  results : Array[Result[T, WorkflowError]],
) -> Array[T] raise WorkflowError {
  let ok = []
  for result in results {
    match result {
      Ok(value) => ok.push(value)
      Err(error) => raise error
    }
  }
  ok
}

///|
/// Policy: keep the successes, requiring at least `min_ok` of them
/// (`QuorumNotReached` otherwise). `min_ok=0` is pure best-effort
/// collection; a negative `min_ok` is treated as 0.
pub fn[T] collect_ok(
  results : Array[Result[T, WorkflowError]],
  min_ok? : Int = 0,
) -> Array[T] raise WorkflowError {
  let ok = []
  for result in results {
    if result is Ok(value) {
      ok.push(value)
    }
  }
  if ok.length() < min_ok {
    raise QuorumNotReached(need=min_ok, ok=ok.length(), of=results.length())
  }
  ok
}

///|
/// Policy: k-of-n agreement gates (adversarial verification, vote panels):
/// at least `need` slots must have succeeded.
pub fn[T] quorum(
  results : Array[Result[T, WorkflowError]],
  need~ : Int,
) -> Array[T] raise WorkflowError {
  collect_ok(results, min_ok=need)
}