///|
fn is_declared_target(
  routes : ReadOnlyArray[DeclaredRoute],
  target : NodeId,
) -> Bool {
  for route in routes {
    if route.target == target {
      return true
    }
  }
  false
}

///|
fn attach_cleanup(primary : Error, cleanup : Error) -> Error {
  match primary {
    ResourceCleanupFailed(failure~) =>
      ResourceCleanupFailed(failure=RunFailure::{
        primary: failure.primary,
        cleanup: ReadOnlyArray::from_array([..failure.cleanup, cleanup]),
      })
    _ =>
      ResourceCleanupFailed(failure=RunFailure::{ primary, cleanup: [cleanup] })
  }
}

///|
async fn[S, P] execute_node(
  node : Node[S, P],
  context : NodeContext,
  state : S,
  timeout_ms : Int?,
) -> NodeOutput[P] {
  let execute = async fn() {
    (node.execute)(context, state) catch {
      error if @async.is_being_cancelled() ||
        @async.is_cancellation_error(error) => raise error
      error =>
        raise GraphRuntimeError::NodeFailed(
          node_id=context.node_id,
          step=context.step,
          cause=error,
        )
    }
  }
  match timeout_ms {
    Some(timeout_ms) =>
      @async.with_timeout(
        timeout_ms,
        execute,
        error=NodeTimedOut(
          node_id=context.node_id,
          step=context.step,
          timeout_ms~,
        ),
      )
    None => execute()
  }
}

///|
async fn[S, P] run_sequential(
  runtime : GraphRuntime[S, P],
  initial_state : S,
  options : RunOptions,
  run_id : RunId,
  group : @async.TaskGroup[Unit],
  resources : ResourceStore,
) -> RunResult[S] {
  let mut state = initial_state
  let mut current = runtime.graph.entry()
  let mut steps = 0
  while true {
    if steps >= options.max_steps {
      raise StepLimitExceeded(limit=options.max_steps)
    }
    steps = steps + 1
    let node = runtime.graph.get_node(current)
    let context = NodeContext::NodeContext(
      run_id,
      current,
      steps,
      group,
      events=runtime.events,
      resources~,
      deadline_ms?=options.node_timeout_ms.map(Int64::from_int),
    )
    runtime.events.try_emit(NodeStarted(run_id~, node_id=current, step=steps))
    let mut output : NodeOutput[P]? = None
    let mut primary : Error? = None
    try {
      output = Some(execute_node(node, context, state, options.node_timeout_ms))
    } catch {
      error if @async.is_being_cancelled() ||
        @async.is_cancellation_error(error) => primary = Some(error)
      error => {
        runtime.events.try_emit(
          NodeFailed(run_id~, node_id=current, step=steps, cause=error),
        )
        primary = Some(error)
      }
    }
    let cleanup = try {
      resources.release_node(current)
      None
    } catch {
      error => Some(error)
    }
    match (primary, cleanup) {
      (Some(error), Some(cleanup_error)) =>
        raise attach_cleanup(error, cleanup_error)
      (Some(error), None) => raise error
      (None, Some(cleanup_error)) =>
        raise ResourceCleanupFailed(failure=RunFailure::{
          primary: cleanup_error,
          cleanup: [],
        })
      (None, None) => ()
    }
    let output = match output {
      Some(output) => output
      None => abort("node attempt completed without output or error")
    }
    let completion = NodeCompletion::{ node_id: current, value: output.value }
    runtime.events.try_emit(
      NodeCompleted(run_id~, node_id=current, step=steps, completion~),
    )
    match output.patch {
      Some(patch) => {
        state = (runtime.graph.reducer.apply)(state, patch) catch {
          error => raise ReduceFailed(node_id=current, step=steps, cause=error)
        }
        runtime.events.try_emit(
          StateUpdated(run_id~, node_id=current, step=steps),
        )
      }
      None => ()
    }
    let router = runtime.graph.get_router(current)
    let route = (router.evaluate)(state, completion) catch {
      error => raise RouteFailed(node_id=current, step=steps, cause=error)
    }
    runtime.events.try_emit(RouteSelected(run_id~, from=current, route~))
    match route {
      To(target) => {
        guard is_declared_target(router.declared_routes, target) else {
          raise RouteContractViolated(from=current, to=target)
        }
        current = target
      }
      End => return RunResult::{ run_id, final_state: state, steps }
      Fail(message) => raise ExplicitFailure(node_id=current, message~)
    }
  } nobreak {
    abort("sequential runtime loop ended unexpectedly")
  }
}

///|
pub async fn[S, P] GraphRuntime::invoke(
  self : GraphRuntime[S, P],
  initial_state : S,
  options? : RunOptions = RunOptions(),
) -> RunResult[S] {
  let run_id = self.fresh_run_id()
  let stored_result : Ref[RunResult[S]?] = Ref(None)
  @async.with_task_group() <| group => {
    let resources = ResourceStore::ResourceStore()
    self.events.try_emit(RunStarted(run_id))
    let mut result : RunResult[S]? = None
    let mut primary : Error? = None
    let mut cancelled : Error? = None
    try {
      result = Some(
        run_sequential(self, initial_state, options, run_id, group, resources),
      )
    } catch {
      error if @async.is_being_cancelled() ||
        @async.is_cancellation_error(error) => cancelled = Some(error)
      error => primary = Some(error)
    }
    let cleanup = try {
      resources.finalize(options.cleanup_timeout_ms)
      None
    } catch {
      error => Some(error)
    }
    match cancelled {
      Some(error) => {
        self.events.try_emit(RunCancelled(run_id))
        raise error
      }
      None => ()
    }
    match (primary, cleanup) {
      (Some(error), Some(cleanup_error)) => {
        let failure = attach_cleanup(error, cleanup_error)
        self.events.try_emit(RunFailed(run_id~, cause=failure))
        raise failure
      }
      (Some(error), None) => {
        self.events.try_emit(RunFailed(run_id~, cause=error))
        raise error
      }
      (None, Some(cleanup_error)) => {
        let failure : Error = ResourceCleanupFailed(failure=RunFailure::{
          primary: cleanup_error,
          cleanup: [],
        })
        self.events.try_emit(RunFailed(run_id~, cause=failure))
        raise failure
      }
      (None, None) =>
        match result {
          Some(result) => {
            self.events.try_emit(RunCompleted(run_id~, steps=result.steps))
            stored_result.val = Some(result)
          }
          None => abort("run completed without result or error")
        }
    }
  }
  match stored_result.val {
    Some(result) => result
    None => abort("task group completed without run result")
  }
}