///|
/// Task ownership and cancellation on an existing host TaskGroup.
///
/// Supervisor is separate from Fuwaroid message processing. It creates no
/// runtime and does not detach tasks from the host group.
pub struct Supervisor[X] {
  priv spawn_task : (async () -> X) -> @async.Task[X]
  priv state : SupervisorState[X]
}

///|
priv struct SupervisorState[X] {
  mut next_id : Int
  mut lifecycle : SupervisorLifecycle
  live : Array[SupervisedTask[X]]
}

///|
pub struct SupervisedTask[X] {
  priv id : Int
  priv label : String
  priv mut status : SupervisedTaskStatus
  priv mut error_text : String?
  priv mut task : @async.Task[X]?
}

///|
/// Bind to a host TaskGroup of any result type.
pub fn[G, X] Supervisor::Supervisor(
  group~ : @async.TaskGroup[G],
) -> Supervisor[X] {
  {
    spawn_task: fn(worker) { group.spawn(no_wait=true, worker) },
    state: { next_id: 1, lifecycle: Open, live: [], },
  }
}

///|
fn[X] supervisor_reap(
  state : SupervisorState[X],
  task : SupervisedTask[X],
) -> Unit {
  for i in 0.. Unit {
  if supervisor_status_transition(task.status, status) {
    task.status = status
    task.error_text = text
    supervisor_reap(state, task)
  }
}

///|
async fn[X] supervised_run(
  state : SupervisorState[X],
  task : SupervisedTask[X],
  f : async () -> X,
) -> X {
  let outcome : Result[X?, Error] = Ok(@async.handle_cancellation(f)) catch {
    error => Err(error)
  }
  match outcome {
    Ok(Some(value)) => {
      supervisor_record_terminal(state, task, Completed)
      value
    }
    Ok(None) => {
      supervisor_record_terminal(state, task, Cancelled)
      // Cancellation is sticky in async 0.22.x. Re-enter an unprotected
      // cancellation point so the underlying task remains Cancelled.
      @async.pause()
      abort("fuwaroid: cancelled supervisee resumed unexpectedly")
    }
    Err(original) => {
      supervisor_record_terminal(state, task, Failed, text=original.to_string())
      raise original
    }
  }
}

///|
/// Spawn a supervised child task. Admission is refused once shutdown starts.
pub fn[X] Supervisor::spawn(
  self : Supervisor[X],
  f : async () -> X,
  label? : String = "",
) -> Result[SupervisedTask[X], SpawnRefusal] {
  if !(self.state.lifecycle is Open) {
    return Err(SupervisorClosed)
  }
  let id = self.state.next_id
  self.state.next_id = id + 1
  let handle : SupervisedTask[X] = {
    id,
    label,
    status: Running,
    error_text: None,
    task: None,
  }
  // Register before spawn because the child may run before spawn returns.
  self.state.live.push(handle)
  let task = (self.spawn_task)(() => supervised_run(self.state, handle, f))
  handle.task = Some(task)
  Ok(handle)
}

///|
/// Request cooperative cancellation. Returning does not imply termination.
pub fn[X] SupervisedTask::cancel(self : SupervisedTask[X]) -> Unit {
  match self.status {
    Running => {
      self.status = Cancelling
      match self.task {
        Some(task) => task.cancel()
        None => abort("fuwaroid: supervisee task missing during cancel")
      }
    }
    Cancelling =>
      match self.task {
        Some(task) => task.cancel()
        None => abort("fuwaroid: supervisee task missing during cancel")
      }
    Completed | Cancelled | Failed => ()
  }
}

///|
/// Wait using the underlying Task semantics.
///
/// A cancelled target raises @async.TaskCancelled. Ordinary worker failures
/// keep their original Error identity. Caller cancellation propagates.
pub async fn[X] SupervisedTask::wait(self : SupervisedTask[X]) -> X {
  match self.task {
    Some(task) => task.wait()
    None => abort("fuwaroid: supervisee task missing during wait")
  }
}

///|
pub fn[X] SupervisedTask::snapshot(
  self : SupervisedTask[X],
) -> SupervisedTaskSnapshot {
  {
    id: self.id,
    label: self.label,
    status: self.status,
    error_text: self.error_text,
  }
}

///|
pub fn[X] Supervisor::snapshot(self : Supervisor[X]) -> SupervisorSnapshot {
  {
    lifecycle: self.state.lifecycle,
    tasks: self.state.live.map(task => task.snapshot()),
  }
}