///|
/// Executes a closure with batched input updates.
///
/// All `Input::set` calls inside the closure are deferred. At batch end,
/// a single revision bump occurs for all changes. This provides:
///
/// - **Atomicity**: Memos see either all changes or none
/// - **Efficiency**: One verification pass instead of many
/// - **Revert detection**: Setting an input back to its original value is a no-op
///
/// Batches can be nested. Only the outermost batch commits changes.
///
/// # Parameters
///
/// - `f`: The closure to execute
///   If `f` raises an error, pending writes in this batch are rolled back and
///   the error is re-raised to the caller.
///
/// # Example
///
/// ```moonbit nocheck
/// rt.batch(() => {
///   x.set(10)
///   y.set(20)
///   z.set(30)
/// })
/// // Single revision bump for all three changes
/// ```
///
/// # Revert Detection
///
/// ```moonbit nocheck
/// rt.batch(() => {
///   x.set(5)   // Change from 0 to 5
///   x.set(0)   // Change back to 0
/// })
/// // No revision bump — net change is zero
/// ```
///
/// # Abort Behavior
///
/// MoonBit `abort()` is not catchable. If `f` aborts (rather than raises),
/// cleanup cannot run and runtime state may be left inconsistent.
pub fn Runtime::batch(self : Runtime, f : () -> Unit raise?) -> Unit raise? {
  self.core.batch.depth = self.core.batch.depth + 1
  self.core.batch.frames.push(BatchFrame::new())
  f() catch {
    e => {
      self.core.batch.depth = self.core.batch.depth - 1
      if self.core.batch.depth < 0 {
        abort("batch_depth underflow: batch() nesting is unbalanced")
      }
      self.rollback_current_batch_frame()
      raise e
    }
  }
  self.core.batch.depth = self.core.batch.depth - 1
  // Defense-in-depth: batch_depth is priv, but guard future modifications to mutation sites
  if self.core.batch.depth < 0 {
    abort("batch_depth underflow: batch() nesting is unbalanced")
  }
  self.complete_batch_frame_success()
  if self.core.batch.depth == 0 {
    self.commit_batch()
  }
}

///|
/// Executes a batch and returns raised errors as `Result` instead of re-raising.
///
/// This is a convenience wrapper around `Runtime::batch` for callers that prefer
/// explicit result handling. Like `Runtime::batch`, this captures raised errors
/// only; `abort()` is not recoverable here and still escapes `Runtime::batch_result`.
pub fn Runtime::batch_result(
  self : Runtime,
  f : () -> Unit raise,
) -> Result[Unit, Error] {
  try self.batch(f) catch {
    error => Err(error)
  } noraise {
    _ => Ok(())
  }
}

///|
/// Rolls back only the currently failing batch frame.
fn Runtime::rollback_current_batch_frame(self : Runtime) -> Unit {
  match self.core.batch.frames.pop() {
    Some(frame) => {
      let mut i = frame.undo_entries.length()
      while i > 0 {
        i = i - 1
        (frame.undo_entries[i].rollback)()
      }
      self.recompute_batch_max_durability()
    }
    None => abort("batch frame underflow: batch() nesting is unbalanced")
  }
}

///|
/// Completes the current frame after successful batch execution.
/// - Outermost success: discard undo log.
/// - Nested success: merge child frame entries into parent for outer rollback.
fn Runtime::complete_batch_frame_success(self : Runtime) -> Unit {
  match self.core.batch.frames.pop() {
    Some(frame) => {
      if self.core.batch.frames.length() == 0 {
        return
      }
      let parent = self.core.batch.frames[self.core.batch.frames.length() - 1]
      for entry in frame.undo_entries {
        if !parent.has_undo_for(entry.cell_id) {
          parent.undo_entries.push(entry)
          parent.undo_ids.add(entry.cell_id)
        }
      }
    }
    None => abort("batch frame underflow: batch() nesting is unbalanced")
  }
}

///|
/// Commit pending batch changes. Thin wrapper delegating to the kernel
/// commit_batch coordinator (see `cells/internal/kernel/batch.mbt` for the
/// I4 preservation checklist). Kept private on Runtime so cells/batch.mbt
/// callers + batch_wbtest.mbt can invoke it via `rt.commit_batch()` without
/// spelling out the kernel state refs.
fn Runtime::commit_batch(self : Runtime) -> Unit {
  self.evaluation_strategies.commit_batch(
    self.core,
    self.pull,
    self.push,
    self.datalog,
    self.runtime_evaluation_event_sink(),
  )
  if self.has_pending_events() {
    self.drain_pending_events_if_idle()
  }
}

///|
/// Records an input as having a pending change during a batch.
///
/// Called internally by `Input::set` during batch operations.
///
/// # Parameters
///
/// - `committable`: A `&Committable` wrapping the `PullInputData` with a
///   pending value. `commit_batch` later calls `do_commit()` on it to apply
///   the pending change and `cell_id()` / `durability()` to update metadata.
fn Runtime::record_batch_input(
  self : Runtime,
  committable : &Committable,
) -> Unit {
  self.core.batch.pending.push(committable)
}

///|
/// Registers a rollback action for the current batch frame.
///
/// Only the first rollback registration per `cell_id` in the frame is stored.
/// If not currently in a batch, this is a no-op.
pub fn Runtime::record_batch_rollback(
  self : Runtime,
  cell_id : CellId,
  rollback : () -> Unit,
) -> Unit {
  // Some internal batch-like flows (e.g. commit callbacks) temporarily
  // raise batch_depth without creating an undo frame. In this mode, writes
  // still use batch commit mechanics, but there is no user-visible frame to roll back.
  if self.core.batch.frames.length() == 0 {
    return
  }
  let frame = self.core.batch.frames[self.core.batch.frames.length() - 1]
  if frame.has_undo_for(cell_id) {
    return
  }
  frame.undo_entries.push({ cell_id, rollback })
  frame.undo_ids.add(cell_id)
}

///|
/// Removes one pending input registration by id.
///
/// Called only from rollback closures during batch undo, which is rare.
/// Uses O(n) copy-and-rebuild because `batch_pending` must remain an ordered
/// queue — insertion order determines commit order in `commit_batch`.
fn Runtime::remove_batch_input(self : Runtime, cell_id : CellId) -> Unit {
  let kept : Array[&Committable] = []
  let mut removed = false
  for c in self.core.batch.pending {
    if !removed && c.cell_id() == cell_id {
      removed = true
    } else {
      kept.push(c)
    }
  }
  self.core.batch.pending.clear()
  for c in kept {
    self.core.batch.pending.push(c)
  }
}

///|
/// Recompute max durability from currently pending inputs.
fn Runtime::recompute_batch_max_durability(self : Runtime) -> Unit {
  let mut max_durability : Durability = Low
  for c in self.core.batch.pending {
    let d = c.durability()
    if d > max_durability {
      max_durability = d
    }
  }
  self.core.batch.max_durability = max_durability
}