tenferro-ad 0.8.0

use std::borrow::Cow;
use std::cell::{Cell, RefCell};
use std::cmp::Reverse;
use std::collections::HashMap;
use std::env;
use std::fmt;
use std::marker::PhantomData;
use std::mem::{size_of, size_of_val};
use std::rc::Rc;
#[cfg(test)]
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, MutexGuard, OnceLock, Weak};
use std::time::{Duration, Instant};

use lru::LruCache;
use num_complex::Complex64;

use crate::extension::{
    validate_eager_extension_target, EagerExtensionBackendKind, EagerExtensionTarget,
};

#[cfg(test)]
use computegraph::graph::Graph;
use computegraph::ValueKey;
#[cfg(test)]
use computegraph::ValueRef;
#[cfg(test)]
use tenferro_ops::input_key::TensorInputKey;
use tenferro_ops::{std_tensor_op::StdTensorOp, SymDim, TensorMeta};
use tenferro_runtime::ad_support::{
    analyze_deferred_semantic_trace, compile_ad_source, ones_tensor, RetainedValue,
};
use tenferro_runtime::program::{ProgramValueMetadata, SemanticFingerprint, SemanticProgram};
use tenferro_runtime::{
    CompiledGraph, EngineId, EngineTarget, ErrorPhase, ExecutionContextIdentity, ExecutionEngine,
    ExtensionModule, GraphCompiler, Prepared, Runtime, RuntimeConfigError, RuntimeEpoch,
    TracedTensor,
};
use tenferro_tensor::Engine;
#[cfg(test)]
use tenferro_tensor::TypedTensor;
use tenferro_tensor::{
    AllocationGroup, CacheStats, CompareDir, DType, DescriptorSlot, DotGeneralConfig, GatherConfig,
    GroupError, IntoShapeVec, PadConfig, ScatterConfig, SliceConfig, Tensor, TensorRead,
    TensorScalar, TensorValue, TensorView,
};

use crate::eager_backend::{
    cpu_host_owned_read, engine_kind_registration, unregisterable_engine_error,
};
#[cfg(test)]
use crate::eager_exec::exec_standard_op_on_tensor_reads_in_session;
use crate::eager_exec::{eager_input_promotion_plan, exec_extension_op_on_tensor_reads};
use crate::error::{ContextId, Error, Result};
use crate::metadata::tensor_meta_from_tensor;
use crate::semantic_extension::SemanticExtensionRuleSet;
use crate::traced::{derivative_trace_from_frozen_program, next_input_key};
use crate::transform_cache::{AdTransformCache, AdTransformCacheLimits};

use crate::AdContext;

pub(crate) type GradSlot = Arc<Mutex<Option<Arc<AdValueRecord>>>>;
pub(crate) type WeakGradSlot = Weak<Mutex<Option<Arc<AdValueRecord>>>>;

mod composite;
mod residuals;
pub(crate) use residuals::{finish_residuals, EagerTrace};

#[derive(Debug, Default, Clone)]
struct EagerOpProfileEntry {
    calls: usize,
    total_time: Duration,
}

thread_local! {
    static EAGER_OP_PROFILE_STATE: RefCell<HashMap<&'static str, EagerOpProfileEntry>> =
        RefCell::new(HashMap::new());
    static EAGER_NO_GRAD_DEPTH: Cell<usize> = const { Cell::new(0) };
    static EAGER_CAPTURE_DEPTH: Cell<usize> = const { Cell::new(0) };
    #[cfg(test)]
    static EAGER_OP_PROFILE_ENABLED_OVERRIDE: RefCell<Option<bool>> = const { RefCell::new(None) };
    #[cfg(test)]
    static EAGER_OP_PROFILE_PRINT_EVERY_OVERRIDE: RefCell<Option<Option<usize>>> = const { RefCell::new(None) };
    #[cfg(test)]
    static EAGER_SEMANTIC_VJP_ENABLED_OVERRIDE: RefCell<Option<bool>> = const { RefCell::new(None) };
}

#[cfg(test)]
pub(crate) static EAGER_SEMANTIC_VJP_EXECUTIONS: AtomicUsize = AtomicUsize::new(0);

pub(crate) fn eager_grad_recording_enabled() -> bool {
    EAGER_NO_GRAD_DEPTH.with(|depth| depth.get() == 0)
}

pub(crate) fn eager_capture_active() -> bool {
    EAGER_CAPTURE_DEPTH.with(|depth| depth.get() > 0)
}

fn eager_semantic_vjp_enabled() -> bool {
    #[cfg(test)]
    if let Some(value) = EAGER_SEMANTIC_VJP_ENABLED_OVERRIDE.with(|state| *state.borrow()) {
        return value;
    }

    // Semantic eager VJP/JVP on by default (Unification 7).
    // Set TENFERRO_EAGER_SEMANTIC_VJP=0 to disable.
    static ENABLED: OnceLock<bool> = OnceLock::new();
    *ENABLED.get_or_init(|| env::var("TENFERRO_EAGER_SEMANTIC_VJP").map_or(true, |v| v != "0"))
}

/// Scope guard that temporarily disables eager operation recording.
///
/// Values computed while this guard is alive are concrete eager tensors, but
/// they do not participate in reverse-mode gradient tracking.
///
/// # Examples
///
/// ```
/// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
///
/// let ctx = EagerRuntime::new()?;
/// let x = EagerTensor::requires_grad_in(
///     Tensor::from_vec_col_major(vec![1], vec![2.0_f64]).unwrap(),
///     ctx.clone(),
/// )?;
/// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
///     let _guard = ctx.no_grad();
///     s.mul(&x, &x)
/// })?;
/// assert!(!y.tracks_grad());
/// # Ok::<(), tenferro_ad::Error>(())
/// ```
#[derive(Debug)]
pub struct EagerNoGradGuard {
    active: bool,
    // Thread-local depth guard: must not be Send so it cannot be moved to and
    // dropped on another thread (which would corrupt the creator's depth).
    _not_send: PhantomData<Rc<()>>,
}

impl Drop for EagerNoGradGuard {
    fn drop(&mut self) {
        if !self.active {
            return;
        }
        EAGER_NO_GRAD_DEPTH.with(|depth| {
            depth.set(depth.get().saturating_sub(1));
        });
        self.active = false;
    }
}

/// Scope guard that keeps semantic-trace recording active for untracked
/// intermediates.
///
/// Under active-edge semantics (issue #1665 Def 1), an operation whose inputs
/// are all untracked produces no autograd nodes and drops its semantic trace.
/// Inside this guard, such operations still record their semantic trace, so a
/// later functional JVP/VJP can differentiate with respect to an untracked or
/// detached leaf. This replaces the pre-Def-1 implicit recording.
///
/// # Examples
///
/// ```
/// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
///
/// let ctx = EagerRuntime::new()?;
/// let x = EagerTensor::from_tensor_in(
///     Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0]).unwrap(),
///     ctx.clone(),
/// )?;
/// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
///     let _capture = ctx.capture_trace();
///     s.mul(&x, &x)
/// })?;
/// let seed = EagerTensor::from_tensor_in(
///     Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 1.0]).unwrap(),
///     ctx.clone(),
/// )?;
/// let dx = ctx.vjp(&mut tenferro_cpu::CpuEngine::serial(), &y, &x, &seed)?;
/// assert_eq!(dx.value()?.as_slice::<f64>().unwrap(), &[2.0, 4.0]);
/// # Ok::<(), tenferro_ad::Error>(())
/// ```
#[derive(Debug)]
pub struct EagerTraceCaptureGuard {
    active: bool,
    // Thread-local depth guard: must not be Send so it cannot be moved to and
    // dropped on another thread (which would corrupt the creator's depth).
    _not_send: PhantomData<Rc<()>>,
}

impl Drop for EagerTraceCaptureGuard {
    fn drop(&mut self) {
        if !self.active {
            return;
        }
        EAGER_CAPTURE_DEPTH.with(|depth| {
            depth.set(depth.get().saturating_sub(1));
        });
        self.active = false;
    }
}

pub(crate) fn eager_op_profile_enabled() -> bool {
    #[cfg(test)]
    if let Some(value) = EAGER_OP_PROFILE_ENABLED_OVERRIDE.with(|state| *state.borrow()) {
        return value;
    }

    static ENABLED: OnceLock<bool> = OnceLock::new();
    *ENABLED.get_or_init(|| env::var("TENFERRO_PROFILE_EAGER_OP_AGG").is_ok())
}

pub(crate) fn eager_op_profile_start() -> Option<Instant> {
    eager_op_profile_enabled().then(Instant::now)
}

pub(crate) fn record_eager_op_profile(section: &'static str, elapsed: Duration) {
    if !eager_op_profile_enabled() {
        return;
    }
    EAGER_OP_PROFILE_STATE.with(|state| {
        let mut state = state.borrow_mut();
        let entry = state.entry(section).or_default();
        entry.calls += 1;
        entry.total_time += elapsed;
    });
}

pub(crate) fn profile_eager_op_section<T>(section: &'static str, f: impl FnOnce() -> T) -> T {
    if !eager_op_profile_enabled() {
        return f();
    }
    let started = Instant::now();
    let result = f();
    record_eager_op_profile(section, started.elapsed());
    result
}

pub(crate) fn maybe_print_eager_op_profile() {
    if !eager_op_profile_enabled() {
        return;
    }
    let Some(print_every) = eager_op_profile_print_every() else {
        return;
    };
    if print_every == 0 {
        return;
    }

    let should_print = EAGER_OP_PROFILE_STATE.with(|state| {
        state
            .borrow()
            .get("nary_op.total")
            .is_some_and(|entry| entry.calls % print_every == 0)
    });
    if should_print {
        print_and_reset_eager_op_profile();
    }
}

fn eager_op_profile_print_every() -> Option<usize> {
    #[cfg(test)]
    if let Some(value) = EAGER_OP_PROFILE_PRINT_EVERY_OVERRIDE.with(|state| *state.borrow()) {
        return value;
    }

    env::var("TENFERRO_PROFILE_EAGER_OP_PRINT_EVERY")
        .ok()?
        .parse()
        .ok()
}

pub(crate) fn print_and_reset_eager_op_profile() {
    EAGER_OP_PROFILE_STATE.with(|state| {
        let mut entries: Vec<_> = state
            .borrow()
            .iter()
            .map(|(section, entry)| (*section, entry.clone()))
            .collect();
        state.borrow_mut().clear();
        entries.sort_by_key(|(_, entry)| Reverse(entry.total_time));

        eprintln!("=== tenferro eager op profile ===");
        for (section, entry) in entries {
            let Some(per_call_us) = eager_op_profile_per_call_us(&entry) else {
                continue;
            };
            eprintln!(
                "{section}: calls={} total={:.6}ms per_call={:.3}us",
                entry.calls,
                entry.total_time.as_secs_f64() * 1.0e3,
                per_call_us,
            );
        }
    });
}

fn eager_op_profile_per_call_us(entry: &EagerOpProfileEntry) -> Option<f64> {
    (entry.calls != 0).then(|| entry.total_time.as_secs_f64() * 1.0e6 / entry.calls as f64)
}

fn runtime_config_error(op: &'static str, source: RuntimeConfigError) -> Error {
    Error::runtime_state_source(op, ErrorPhase::Execution, source)
}

fn runtime_state_source<E>(op: &'static str, source: E) -> Error
where
    E: std::error::Error + Send + Sync + 'static,
{
    Error::runtime_state_source(op, ErrorPhase::Execution, source)
}

/// Run `prepared` on `engine` and return owned compact outputs.
pub(crate) fn run_prepared_on(
    runtime: &Runtime,
    engine: &mut dyn ExecutionEngine,
    prepared: &Prepared,
    inputs: &[TensorRead<'_>],
) -> Result<Vec<Tensor>> {
    let outputs = runtime.run(&mut [&mut *engine], prepared, inputs)?;
    outputs
        .into_iter()
        .map(|output| output.into_tensor(&mut *engine).map_err(Error::from))
        .collect()
}

/// Run a session-capable prepared extension executor on `engine`, lending it
/// the engine's extension caches for the call.
pub(crate) fn execute_prepared_extension_on(
    engine: &mut dyn ExecutionEngine,
    executor: &dyn tenferro_runtime::PreparedOperationExecutor,
    inputs: &[TensorRead<'_>],
) -> Result<Vec<Tensor>> {
    with_lent_extension_caches(engine, |engine, caches| {
        executor.execute_in_session(engine, caches, inputs)
    })
}

/// Lend `engine`'s extension caches to `callback` and restore them on every
/// exit path, including unwind.
///
/// A callback panic also marks the engine broken before it is resumed: the
/// caches may be half updated, so the engine must not accept more work.
pub(crate) fn with_lent_extension_caches<R>(
    engine: &mut dyn ExecutionEngine,
    callback: impl FnOnce(&mut dyn ExecutionEngine, &mut tenferro_runtime::ExtensionCacheStore) -> R,
) -> R {
    let mut caches = engine.take_extension_caches();
    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        callback(&mut *engine, &mut caches)
    }));
    engine.restore_extension_caches(caches);
    match result {
        Ok(value) => value,
        Err(payload) => {
            engine.mark_broken();
            std::panic::resume_unwind(payload)
        }
    }
}

/// The eager extension backend kind of an engine target, or `None` for an
/// engine type without eager extension support.
pub(crate) fn eager_extension_backend_kind(
    target: &EngineTarget,
) -> Option<EagerExtensionBackendKind> {
    if target.context_identity() == ExecutionContextIdentity::of::<tenferro_cpu::CpuEngine>() {
        return Some(EagerExtensionBackendKind::Cpu);
    }
    #[cfg(feature = "cuda")]
    if target.context_identity() == ExecutionContextIdentity::of::<tenferro_gpu::cuda::CudaEngine>()
    {
        return Some(EagerExtensionBackendKind::Cuda);
    }
    None
}

/// Stats for caches owned by an [`EagerRuntime`].
///
/// `retained_bytes` fields are logical payload estimates, not process RSS.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct EagerRuntimeCacheStats {
    /// Eager AD transform memoization cache.
    pub ad_transforms: CacheStats,
    /// Prepared eager derivative program cache.
    pub prepared_derivatives: CacheStats,
}

#[cfg(test)]
pub(crate) struct EagerGraphExecution {
    pub(crate) outputs: Vec<Tensor>,
}

/// A read-only value view retained by an eager tensor record.
///
/// The guard borrows the record's allocation group. It never owns a tensor and
/// cannot be converted into a mutable view.
///
/// # Examples
///
/// ```
/// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
///
/// let ctx = EagerRuntime::new()?;
/// let value = EagerTensor::from_tensor_in(
///     Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0])?,
///     ctx,
/// )?;
/// let view = value.value()?;
/// assert_eq!(view.shape(), &[2]);
/// # Ok::<(), tenferro_ad::Error>(())
/// ```
#[derive(Debug)]
pub struct ValueGuard<'a> {
    view: TensorView<'a>,
}

impl<'a> ValueGuard<'a> {
    /// Return the scalar dtype of the retained value.
    pub fn dtype(&self) -> DType {
        self.view.dtype()
    }

    /// Return the logical shape of the retained value.
    pub fn shape(&self) -> &[usize] {
        self.view.shape()
    }

    /// Borrow the dtype-erased tensor view.
    pub fn as_tensor_view(&self) -> &TensorView<'_> {
        &self.view
    }

    /// Borrow compact host bytes through the tensor's explicit scalar type.
    ///
    /// Backend-resident values return the backend's typed host-access error;
    /// this method does not download storage implicitly.
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_tensor::ValidationError::DTypeMismatch`] when
    /// `T` does not match the view dtype, [`tenferro_tensor::ValidationError::NonContiguousViewAsSlice`]
    /// for a non-contiguous view, or [`tenferro_tensor::Error::HostAccess`]
    /// when backend storage cannot be mapped as a host slice.
    pub fn as_slice<T: TensorScalar>(&self) -> tenferro_tensor::Result<&'a [T]> {
        self.view.as_slice()
    }

    fn duplicate_host_tensor(&self) -> tenferro_tensor::Result<Tensor> {
        match &self.view {
            TensorView::F32(view) => {
                <f32 as TensorScalar>::into_tensor(view.shape().to_vec(), view.as_slice()?.to_vec())
            }
            TensorView::F64(view) => {
                <f64 as TensorScalar>::into_tensor(view.shape().to_vec(), view.as_slice()?.to_vec())
            }
            TensorView::I32(view) => {
                <i32 as TensorScalar>::into_tensor(view.shape().to_vec(), view.as_slice()?.to_vec())
            }
            TensorView::I64(view) => {
                <i64 as TensorScalar>::into_tensor(view.shape().to_vec(), view.as_slice()?.to_vec())
            }
            TensorView::Bool(view) => <bool as TensorScalar>::into_tensor(
                view.shape().to_vec(),
                view.as_slice()?.to_vec(),
            ),
            TensorView::C32(view) => <num_complex::Complex32 as TensorScalar>::into_tensor(
                view.shape().to_vec(),
                view.as_slice()?.to_vec(),
            ),
            TensorView::C64(view) => <num_complex::Complex64 as TensorScalar>::into_tensor(
                view.shape().to_vec(),
                view.as_slice()?.to_vec(),
            ),
        }
    }
}

/// Read-only retained gradient value.
///
/// # Examples
///
/// ```
/// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
///
/// let ctx = EagerRuntime::new()?;
/// let x = EagerTensor::requires_grad_in(
///     Tensor::from_vec_col_major(vec![2], vec![2.0_f64, 3.0])?,
///     ctx,
/// )?;
/// let loss = x.runtime().with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
///     let squared = s.mul(&x, &x)?;
///     s.reduce_sum(&squared, Some(&[0]))
/// })?;
/// let _gradients = loss.backward(&mut tenferro_cpu::CpuEngine::serial())?;
/// let gradient = x.grad()?.expect("tracked leaf has a gradient");
/// assert_eq!(gradient.shape(), &[2]);
/// # Ok::<(), tenferro_ad::Error>(())
/// ```
#[derive(Clone, Debug)]
pub struct GradientValue {
    record: Arc<AdValueRecord>,
}

impl GradientValue {
    /// Return the scalar dtype of the gradient.
    pub fn dtype(&self) -> DType {
        self.record.dtype()
    }

    /// Return the logical shape of the gradient.
    pub fn shape(&self) -> &[usize] {
        self.record.shape()
    }

    /// Borrow the gradient's value guard.
    ///
    /// # Errors
    ///
    /// Returns [`Error::RuntimeState`] when the retained gradient record is
    /// unavailable or its allocation-group descriptor is invalid.
    pub fn value(&self) -> Result<ValueGuard<'_>> {
        self.record.value("GradientValue::value")
    }

    /// Borrow the gradient as a dtype-erased read target.
    ///
    /// # Errors
    ///
    /// Returns [`Error::RuntimeState`] when the retained gradient record or
    /// its allocation-group descriptor is unavailable.
    pub fn tensor_read(&self) -> Result<TensorRead<'_>> {
        self.record.tensor_read("GradientValue::tensor_read")
    }

    /// Borrow a compact host slice without downloading backend storage.
    ///
    /// # Errors
    ///
    /// Returns [`Error::RuntimeState`] when the retained value is unavailable,
    /// [`tenferro_tensor::ValidationError::DTypeMismatch`] when `T` does
    /// not match the gradient dtype, or [`tenferro_tensor::Error::HostAccess`]
    /// when backend storage cannot be mapped as a host slice.
    pub fn as_slice<T: TensorScalar>(&self) -> tenferro_tensor::Result<&[T]> {
        self.record
            .value("GradientValue::as_slice")
            .map_err(|error| {
                tenferro_tensor::Error::runtime_state_source("GradientValue::as_slice", error)
            })?
            .as_slice()
    }

    /// Explicitly copy a host-resident gradient into a standalone tensor.
    ///
    /// # Errors
    ///
    /// Returns [`Error::RuntimeState`] when the retained value or execution
    /// session is unavailable, or a typed backend/host-access error when the
    /// gradient cannot be materialized as a contiguous tensor.
    pub fn to_tensor(&self, engine: &mut dyn Engine) -> Result<Tensor> {
        let value = self
            .record
            .value("GradientValue::to_tensor")
            .map_err(|error| {
                Error::runtime_state_source(
                    "GradientValue::to_tensor",
                    ErrorPhase::Execution,
                    error,
                )
            })?;
        match value.duplicate_host_tensor() {
            Ok(tensor) => Ok(tensor),
            Err(_) => {
                let read = self.record.tensor_read("GradientValue::to_tensor")?;
                engine.to_contiguous_read(read).map_err(Error::from)
            }
        }
    }
}

/// Move-only accumulated gradient bundle backed by one allocation group.
///
/// # Examples
///
/// ```
/// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
///
/// let ctx = EagerRuntime::new()?;
/// let x = EagerTensor::requires_grad_in(
///     Tensor::from_vec_col_major(vec![2], vec![2.0_f64, 3.0])?,
///     ctx,
/// )?;
/// let loss = x.runtime().with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
///     let squared = s.mul(&x, &x)?;
///     s.reduce_sum(&squared, Some(&[0]))
/// })?;
/// let gradients = loss.backward(&mut tenferro_cpu::CpuEngine::serial())?;
/// assert!(!gradients.is_empty());
/// # Ok::<(), tenferro_ad::Error>(())
/// ```
#[derive(Debug)]
pub struct Gradients {
    group: AllocationGroup,
    slots: HashMap<ValueKey<StdTensorOp>, DescriptorSlot>,
}

impl Gradients {
    fn from_tensors(tensors: HashMap<ValueKey<StdTensorOp>, Tensor>) -> Result<Self> {
        let (keys, values): (Vec<_>, Vec<_>) = tensors.into_iter().unzip();
        let (group, bindings) = AllocationGroup::from_tensors(values).map_err(|error| {
            Error::runtime_state_source("Gradients::from_tensors", ErrorPhase::Execution, error)
        })?;
        let slots = keys.into_iter().zip(bindings).collect();
        Ok(Self { group, slots })
    }

    /// Return the number of retained gradient descriptors.
    pub fn len(&self) -> usize {
        self.slots.len()
    }

    /// Return whether no gradient was produced.
    pub fn is_empty(&self) -> bool {
        self.slots.is_empty()
    }

    /// Borrow one gradient view by its local value key.
    pub fn grad(&self, key: &ValueKey<StdTensorOp>) -> Option<TensorView<'_>> {
        let slot = self.slots.get(key).copied()?;
        let mut reads = self.group.read_views(std::slice::from_ref(&slot)).ok()?;
        match reads.pop()? {
            TensorRead::View(view) => Some(view),
            TensorRead::Tensor(_) => None,
        }
    }

    /// Consume one gradient owner while leaving the bundle unchanged on failure.
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_tensor::Error::RuntimeState`] when the descriptor is
    /// invalid or its allocation is aliased. A missing key is reported as
    /// `Ok(None)`.
    pub fn take_grad(
        &mut self,
        key: &ValueKey<StdTensorOp>,
    ) -> tenferro_tensor::Result<Option<Tensor>> {
        let Some(&slot) = self.slots.get(key) else {
            return Ok(None);
        };
        let tensor = self.group.take_tensor(slot).map_err(|error| {
            tenferro_tensor::Error::runtime_state_source("Gradients::take_grad", error)
        })?;
        self.slots.remove(key);
        Ok(Some(tensor))
    }
}

/// Error returned when a value cannot be consumed without changing its owner.
///
/// # Examples
///
/// ```
/// use tenferro_ad::{EagerRuntime, EagerTensor, IntoValueError, Tensor};
///
/// let ctx = EagerRuntime::new()?;
/// let value = EagerTensor::from_tensor_in(
///     Tensor::from_vec_col_major(vec![1], vec![1.0_f64])?,
///     ctx,
/// )?;
/// let _shared = value.clone();
/// assert!(matches!(
///     value.into_value(),
///     Err(IntoValueError::NotUnique(_))
/// ));
/// # Ok::<(), tenferro_ad::Error>(())
/// ```
#[derive(Debug)]
pub enum IntoValueError<H> {
    /// Another eager handle, tape record, or checkpoint retains the value.
    NotUnique(H),
    /// Group extraction failed after the handle was uniquely acquired.
    Extract { value: H, error: GroupError },
}

impl<H> std::fmt::Display for IntoValueError<H> {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::NotUnique(_) => formatter.write_str("eager value is retained by another handle"),
            Self::Extract { error, .. } => {
                write!(formatter, "eager value extraction failed: {error}")
            }
        }
    }
}

impl<H: std::fmt::Debug + Send + Sync + 'static> std::error::Error for IntoValueError<H> {}

/// What one direct retention container holds.
// The pooled variant owns an allocation group inline; boxing it would add an
// allocation to every retained value on the hot path.
#[allow(clippy::large_enum_variant)]
#[derive(Debug)]
enum RetentionContainer {
    /// Pooled storage owned through an allocation group and one descriptor slot.
    Pooled {
        group: AllocationGroup,
        slot: DescriptorSlot,
    },
    /// A caller-owned value the runtime retains without taking pool ownership.
    ///
    /// The value returns to its owner when the record drops, and the runtime never
    /// substitutes it for pooled storage. A caller-owned payload has no typed
    /// descriptor view, so only its read and consume paths are available.
    CallerOwned {
        /// Boxed because a tensor value is much larger than the pooled variant.
        tensor: Box<Tensor>,
    },
    /// An untracked result held directly.
    ///
    /// No AD group, residual or gradient will share it, so it needs no
    /// allocation group: the tensor's own storage returns to its pool when the
    /// record drops, and a unique handle hands the tensor back unchanged.
    Owned { tensor: Tensor },
}

/// Read-only descriptor record used by eager handles and the AD registries.
#[derive(Debug)]
pub(crate) struct AdValueRecord {
    container: Arc<RetentionContainer>,
    dtype: DType,
    shape: Box<[usize]>,
}

impl AdValueRecord {
    fn from_group(
        group: AllocationGroup,
        slot: DescriptorSlot,
        dtype: DType,
        shape: Vec<usize>,
    ) -> Arc<Self> {
        Arc::new(Self {
            container: Arc::new(RetentionContainer::Pooled { group, slot }),
            dtype,
            shape: shape.into_boxed_slice(),
        })
    }

    fn from_tensor(tensor: Tensor, op: &'static str) -> Result<Arc<Self>> {
        let dtype = tensor.dtype();
        let shape = tensor.shape().to_vec();
        if matches!(dtype, DType::External(_)) {
            // A caller-owned payload is retained directly: it owns no pooled
            // storage, so there is no group to build and nothing to return to a
            // pool when the record drops.
            return Ok(Arc::new(Self {
                container: Arc::new(RetentionContainer::CallerOwned {
                    tensor: Box::new(tensor),
                }),
                dtype,
                shape: shape.into_boxed_slice(),
            }));
        }
        let (group, bindings) = AllocationGroup::from_tensors(vec![tensor])
            .map_err(|error| Error::runtime_state_source(op, ErrorPhase::Execution, error))?;
        let slot = bindings.first().copied().ok_or_else(|| {
            Error::runtime_state(op, ErrorPhase::Execution, "empty allocation-group binding")
        })?;
        Ok(Self::from_group(group, slot, dtype, shape))
    }

    /// Retain an untracked result without building an allocation group.
    fn from_untracked_tensor(tensor: Tensor, op: &'static str) -> Result<Arc<Self>> {
        if matches!(tensor.dtype(), DType::External(_)) {
            return Self::from_tensor(tensor, op);
        }
        let dtype = tensor.dtype();
        let shape = tensor.shape().to_vec().into_boxed_slice();
        Ok(Arc::new(Self {
            container: Arc::new(RetentionContainer::Owned { tensor }),
            dtype,
            shape,
        }))
    }

    fn tensor_read(&self, op: &'static str) -> Result<TensorRead<'_>> {
        match self.container.as_ref() {
            RetentionContainer::Pooled { group, slot } => group.read_view(*slot).map_err(|error| {
                // One slot, so no `Vec` and no empty-result case: the single-slot
                // accessor is the whole request (#2079).
                Error::runtime_state_source(op, ErrorPhase::Execution, error)
            }),
            RetentionContainer::CallerOwned { tensor } => Ok(TensorRead::from_tensor(tensor)),
            RetentionContainer::Owned { tensor } => Ok(TensorRead::from_tensor(tensor)),
        }
    }

    fn value(&self, op: &'static str) -> Result<ValueGuard<'_>> {
        if let RetentionContainer::Owned { tensor } = self.container.as_ref() {
            // A preset-dtype owned tensor always has a typed borrowed view.
            return Ok(ValueGuard {
                view: TensorRead::from_tensor(tensor).tensor_view(),
            });
        }
        match self.tensor_read(op)? {
            TensorRead::View(view) => Ok(ValueGuard { view }),
            // A caller-owned payload has no typed descriptor view, so a path that
            // needs one fails explicitly instead of borrowing the payload as bytes.
            TensorRead::Tensor(_) => Err(Error::runtime_state(
                op,
                ErrorPhase::Execution,
                "allocation-group value did not produce a borrowed descriptor view",
            )),
        }
    }

    fn dtype(&self) -> DType {
        self.dtype
    }

    fn shape(&self) -> &[usize] {
        &self.shape
    }
}

/// Shared eager execution context for tensors on a backend.
///
/// Reusing one context lets eager tensors share backend state, extension
/// runtime caches, and gradient storage across a computation.
///
/// # Examples
///
/// ```
/// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
///
/// let ctx = EagerRuntime::new()?;
/// let x = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![1], vec![1.0_f64]).unwrap(), ctx.clone()).unwrap();
/// let y = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![1], vec![2.0_f64]).unwrap(), ctx.clone()).unwrap();
/// let z = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| session.add(&x, &y)).unwrap();
///
/// assert_eq!(z.value().unwrap().as_slice::<f64>().unwrap(), &[3.0]);
/// # Ok::<(), tenferro_ad::Error>(())
/// ```
pub struct EagerRuntime {
    id: ContextId,
    /// Prepared-program runtime. Engine kinds are registered on first use;
    /// the runtime never stores an engine.
    runtime: Runtime,
    /// Serializes runtime reconfiguration (engine-kind registration and
    /// extension-module installation).
    extension_install_lock: Mutex<()>,
    semantic_extension_rules: SemanticExtensionRuleSet,
    grad_slots: Mutex<HashMap<ValueKey<StdTensorOp>, WeakGradSlot>>,
    value_records: Mutex<HashMap<ValueKey<StdTensorOp>, Weak<EagerTensorRecord>>>,
    ad_transform_cache: Arc<AdTransformCache>,
    /// S2: prepared derivative programs keyed by semantic structure, wrt input,
    /// and concrete bound input metadata. Avoids re-running freeze+AD
    /// transform+compile_frozen on warm structure hits.
    prepared_derivative_cache: Mutex<PreparedDerivativeCache>,
}

/// An eager runtime and its borrowed backend session for one execution boundary.
///
/// Obtain this only through [`EagerRuntime::with_eager_session`]. It rejects
/// tensors from another eager runtime even if both runtimes use the same backend
/// type, and it cannot escape the boundary closure.
///
/// # Examples
///
/// ```rust
/// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
///
/// let ctx = EagerRuntime::new()?;
/// let x = EagerTensor::from_tensor_in(
///     Tensor::from_vec_col_major(vec![1], vec![3.0_f64])?, ctx.clone(),
/// )?;
/// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| session.neg(&x))?;
/// assert_eq!(y.value()?.as_slice::<f64>()?, &[-3.0]);
/// # Ok::<(), tenferro_ad::Error>(())
/// ```
///
/// The borrowed session cannot escape its execution boundary:
///
/// ```compile_fail
/// use tenferro_ad::EagerRuntime;
/// let ctx = EagerRuntime::new().unwrap();
/// let escaped = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| session).unwrap();
/// let _ = escaped;
/// ```
pub struct EagerSession<'a> {
    runtime: &'a Arc<EagerRuntime>,
    engine: &'a mut dyn ExecutionEngine,
}

impl fmt::Debug for EagerSession<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("EagerSession")
            .field("runtime_id", &self.runtime.id())
            .finish_non_exhaustive()
    }
}

impl EagerSession<'_> {
    /// Negate an eager tensor inside the caller's execution boundary.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::from_tensor_in(
    ///     Tensor::from_vec_col_major(vec![1], vec![4.0_f64])?, ctx.clone(),
    /// )?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| session.neg(&x))?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[-4.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for a tensor from another runtime,
    /// or a typed eager/backend error from the selected operation.
    pub fn neg(&mut self, input: &EagerTensor) -> Result<EagerTensor> {
        self.run_unary(input, StdTensorOp::Neg)
    }

    /// Compute the elementwise exponential inside this eager session.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![1], vec![0.0_f64])?, ctx.clone())?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| session.exp(&x))?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[1.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for a foreign runtime, or a typed
    /// unsupported/backend error for the input dtype.
    pub fn exp(&mut self, input: &EagerTensor) -> Result<EagerTensor> {
        self.run_unary(input, StdTensorOp::Exp)
    }

    /// Compute the elementwise absolute value inside this eager session.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![1], vec![-2.0_f64])?, ctx.clone())?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| session.abs(&x))?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[2.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for a foreign runtime, or a typed
    /// unsupported/backend error for the input dtype.
    pub fn abs(&mut self, input: &EagerTensor) -> Result<EagerTensor> {
        self.run_unary(input, StdTensorOp::Abs)
    }

    /// Compute the elementwise conjugate inside this eager session.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![1], vec![2.0_f64])?, ctx.clone())?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| session.conj(&x))?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[2.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for a foreign runtime, or a typed
    /// unsupported/backend error for the input dtype.
    pub fn conj(&mut self, input: &EagerTensor) -> Result<EagerTensor> {
        self.run_unary(input, StdTensorOp::Conj)
    }

    /// Compute the elementwise sign on this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![1], vec![-2.0_f64])?)?;
    ///     s.sign(&x)
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[-1.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, unsupported-dtype, or backend error.
    pub fn sign(&mut self, input: &EagerTensor) -> Result<EagerTensor> {
        self.run_unary(input, StdTensorOp::Sign)
    }

    /// Compute the elementwise natural logarithm on this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![1], vec![1.0_f64])?)?;
    ///     s.log(&x)
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[0.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, unsupported-dtype, or backend error.
    pub fn log(&mut self, input: &EagerTensor) -> Result<EagerTensor> {
        self.run_unary(input, StdTensorOp::Log)
    }

    /// Compute the elementwise square root on this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![1], vec![4.0_f64])?)?;
    ///     s.sqrt(&x)
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[2.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, unsupported-dtype, or backend error.
    pub fn sqrt(&mut self, input: &EagerTensor) -> Result<EagerTensor> {
        self.run_unary(input, StdTensorOp::Sqrt)
    }

    /// Compute the elementwise reciprocal square root on this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![1], vec![4.0_f64])?)?;
    ///     s.rsqrt(&x)
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[0.5]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, unsupported-dtype, or backend error.
    pub fn rsqrt(&mut self, input: &EagerTensor) -> Result<EagerTensor> {
        self.run_unary(input, StdTensorOp::Rsqrt)
    }

    /// Compute the elementwise sine on this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![1], vec![0.0_f64])?)?;
    ///     s.sin(&x)
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[0.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, unsupported-dtype, or backend error.
    pub fn sin(&mut self, input: &EagerTensor) -> Result<EagerTensor> {
        self.run_unary(input, StdTensorOp::Sin)
    }

    /// Compute the elementwise cosine on this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![1], vec![0.0_f64])?)?;
    ///     s.cos(&x)
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[1.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, unsupported-dtype, or backend error.
    pub fn cos(&mut self, input: &EagerTensor) -> Result<EagerTensor> {
        self.run_unary(input, StdTensorOp::Cos)
    }

    /// Compute the elementwise hyperbolic tangent on this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![1], vec![0.0_f64])?)?;
    ///     s.tanh(&x)
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[0.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, unsupported-dtype, or backend error.
    pub fn tanh(&mut self, input: &EagerTensor) -> Result<EagerTensor> {
        self.run_unary(input, StdTensorOp::Tanh)
    }

    /// Compute `exp(x) - 1` elementwise on this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![1], vec![0.0_f64])?)?;
    ///     s.expm1(&x)
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[0.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, unsupported-dtype, or backend error.
    pub fn expm1(&mut self, input: &EagerTensor) -> Result<EagerTensor> {
        self.run_unary(input, StdTensorOp::Expm1)
    }

    /// Compute `log(1 + x)` elementwise on this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![1], vec![0.0_f64])?)?;
    ///     s.log1p(&x)
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[0.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, unsupported-dtype, or backend error.
    pub fn log1p(&mut self, input: &EagerTensor) -> Result<EagerTensor> {
        self.run_unary(input, StdTensorOp::Log1p)
    }

    /// Compute the error function `erf(x)` elementwise on this borrowed session.
    ///
    /// Defined for real `F32`/`F64` tensors; `erf(+-0) = +-0`,
    /// `erf(+-inf) = +-1`, and `NaN` stays `NaN`. The derivative is
    /// `2/sqrt(pi) * exp(-x^2)`.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![2], vec![0.0_f64, 1.0])?)?;
    ///     s.erf(&x)
    /// })?;
    /// let y = y.value()?;
    /// let y = y.as_slice::<f64>()?;
    /// assert_eq!(y[0], 0.0);
    /// assert!((y[1] - 0.842_700_792_949_714_9).abs() < 1.0e-15);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime error, a typed unsupported-dtype error
    /// for complex, integer, or `Bool` input, or a backend error.
    pub fn erf(&mut self, input: &EagerTensor) -> Result<EagerTensor> {
        self.run_unary(input, StdTensorOp::Erf)
    }

    /// Convert a tensor under the checked dtype-promotion lattice.
    /// Use [`Self::cast`] for intentional lossy projection.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{DType, EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let converted = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| {
    ///     let x = session.constant_from(Tensor::from_vec_col_major(vec![1], vec![2.0_f64])?)?;
    ///     session.convert(&x, DType::C64)
    /// })?;
    /// assert_eq!(converted.dtype(), DType::C64);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for a foreign runtime or a typed
    /// unsupported dtype conversion/backend error.
    pub fn convert(&mut self, input: &EagerTensor, to: DType) -> Result<EagerTensor> {
        self.ensure_runtime(input)?;
        tenferro_tensor::validate::validate_convert_dtype(
            "EagerTensor::convert",
            input.dtype(),
            to,
        )
        .map_err(Error::TensorRuntime)?;
        self.cast(input, to)
    }

    /// Cast a tensor to a dtype, permitting explicitly lossy projections.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{DType, EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let casted = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| {
    ///     let x = session.constant_from(Tensor::from_vec_col_major(vec![1], vec![2.8_f64])?)?;
    ///     session.cast(&x, DType::I32)
    /// })?;
    /// assert_eq!(casted.value()?.as_slice::<i32>()?, &[2]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for a foreign runtime, or a typed
    /// unsupported projection/backend error.
    pub fn cast(&mut self, input: &EagerTensor, to: DType) -> Result<EagerTensor> {
        self.run_unary(
            input,
            StdTensorOp::Convert {
                from: input.dtype(),
                to,
            },
        )
    }

    /// Permute the axes of an eager tensor while preserving independent ownership.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let copied = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| {
    ///     let x = session.constant_from(Tensor::from_vec_col_major(vec![2, 2], vec![1.0_f64, 2.0, 3.0, 4.0])?)?;
    ///     let y = session.transpose(&x, &[1, 0])?;
    ///     session.duplicate_value(&y)
    /// })?;
    /// assert_eq!(copied.as_slice::<f64>()?, &[1.0, 3.0, 2.0, 4.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for a foreign runtime, or a typed
    /// axis/backend error for an invalid permutation or copy.
    pub fn transpose(&mut self, input: &EagerTensor, perm: &[usize]) -> Result<EagerTensor> {
        self.ensure_runtime(input)?;
        input.transpose_in_session(perm, &mut *self.engine)
    }

    /// Reshape an eager tensor while retaining a separate result owner.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0])?, ctx.clone())?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| session.reshape(&x, [1, 2]))?;
    /// assert_eq!(y.shape(), &[1, 2]);
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[1.0, 2.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for a foreign runtime, or a typed
    /// validation/backend error when the target shape is incompatible.
    pub fn reshape(
        &mut self,
        input: &EagerTensor,
        shape: impl IntoShapeVec,
    ) -> Result<EagerTensor> {
        self.ensure_runtime(input)?;
        input.reshape_in_session(&shape.into_shape_vec(), &mut *self.engine)
    }

    /// Slice an eager tensor with explicit start, limit, and stride per axis.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, SliceConfig, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| {
    ///     let x = session.constant_from(Tensor::from_vec_col_major(vec![4], vec![1.0_f64, 2.0, 3.0, 4.0])?)?;
    ///     session.slice(&x, SliceConfig { starts: vec![1], limits: vec![3], strides: vec![1] })
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[2.0, 3.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for a foreign runtime, or a typed
    /// axis/stride/backend error for an invalid slice or copy.
    pub fn slice(&mut self, input: &EagerTensor, config: SliceConfig) -> Result<EagerTensor> {
        self.ensure_runtime(input)?;
        input.slice_in_session(config, &mut *self.engine)
    }

    /// Broadcast an eager tensor into a larger shape on this session.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0])?, ctx.clone())?;
    /// let copy = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| {
    ///     let y = session.broadcast_in_dim(&x, &[2, 2], &[0])?;
    ///     session.duplicate_value(&y)
    /// })?;
    /// assert_eq!(copy.as_slice::<f64>()?, &[1.0, 2.0, 1.0, 2.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for a foreign runtime, or a typed
    /// validation/backend error for an invalid broadcast mapping.
    pub fn broadcast_in_dim(
        &mut self,
        input: &EagerTensor,
        shape: &[usize],
        dims: &[usize],
    ) -> Result<EagerTensor> {
        self.ensure_runtime(input)?;
        input.broadcast_in_dim_in_session(shape, dims, &mut *self.engine)
    }

    /// Keep the lower triangle of a matrix on this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let lower = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let matrix = s.constant_from(Tensor::from_vec_col_major(vec![2, 2], vec![1.0_f64, 2.0, 3.0, 4.0])?)?;
    ///     s.tril(&matrix, 0)
    /// })?;
    /// assert_eq!(lower.value()?.as_slice::<f64>()?, &[1.0, 2.0, 0.0, 4.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, rank, unsupported-dtype, or backend error.
    pub fn tril(&mut self, input: &EagerTensor, k: i64) -> Result<EagerTensor> {
        self.run_unary(input, StdTensorOp::Tril { k })
    }

    /// Keep the upper triangle of a matrix on this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let upper = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let matrix = s.constant_from(Tensor::from_vec_col_major(vec![2, 2], vec![1.0_f64, 2.0, 3.0, 4.0])?)?;
    ///     s.triu(&matrix, 0)
    /// })?;
    /// assert_eq!(upper.value()?.as_slice::<f64>()?, &[1.0, 0.0, 3.0, 4.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, rank, unsupported-dtype, or backend error.
    pub fn triu(&mut self, input: &EagerTensor, k: i64) -> Result<EagerTensor> {
        self.run_unary(input, StdTensorOp::Triu { k })
    }

    /// Pad an eager tensor with zeros on this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, PadConfig, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let padded = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0])?)?;
    ///     s.pad(&x, PadConfig {
    ///         edge_padding_low: vec![1],
    ///         edge_padding_high: vec![1],
    ///         interior_padding: vec![1],
    ///     })
    /// })?;
    /// assert_eq!(padded.value()?.as_slice::<f64>()?, &[0.0, 1.0, 0.0, 2.0, 0.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns [`Error::ContextMismatch`] when an input belongs to another eager
    /// runtime, a validation error with
    /// `ValidationError::InvalidArgument` for a padding configuration whose
    /// length or extents do not match the input rank, or
    /// [`Error::TensorRuntime`] for a typed backend failure.
    pub fn pad(&mut self, input: &EagerTensor, config: PadConfig) -> Result<EagerTensor> {
        self.run_unary(input, StdTensorOp::Pad(config))
    }

    /// Reverse the elements along selected axes on this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let reversed = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![3], vec![1.0_f64, 2.0, 3.0])?)?;
    ///     s.reverse(&x, &[0])
    /// })?;
    /// assert_eq!(reversed.value()?.as_slice::<f64>()?, &[3.0, 2.0, 1.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, invalid-axis, or backend error.
    pub fn reverse(&mut self, input: &EagerTensor, axes: &[usize]) -> Result<EagerTensor> {
        self.ensure_runtime(input)?;
        crate::eager_ops::validate_eager_axes("EagerSession::reverse", input.shape().len(), axes)?;
        self.run_unary(
            input,
            StdTensorOp::Reverse {
                axes: axes.to_vec(),
            },
        )
    }

    /// Slice an eager tensor using runtime start indices in this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let selected = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![4], vec![1.0_f64, 2.0, 3.0, 4.0])?)?;
    ///     let starts = s.constant_from(Tensor::from_vec_col_major(vec![1], vec![1_i64])?)?;
    ///     s.dynamic_slice(&x, &starts, &[2])
    /// })?;
    /// assert_eq!(selected.value()?.as_slice::<f64>()?, &[2.0, 3.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, invalid-index or slice-shape, or backend error.
    pub fn dynamic_slice(
        &mut self,
        input: &EagerTensor,
        starts: &EagerTensor,
        sizes: &[usize],
    ) -> Result<EagerTensor> {
        self.ensure_runtime(input)?;
        self.ensure_runtime(starts)?;
        EagerTensor::nary_op_in_session(
            &[input, starts],
            StdTensorOp::DynamicSlice {
                slice_sizes: sizes.to_vec(),
            },
            &mut *self.engine,
        )
    }

    /// Gather elements of an eager tensor in this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, GatherConfig, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let result = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![3], vec![10.0_f64, 20.0, 30.0])?)?;
    ///     let indices = s.constant_from(Tensor::from_vec_col_major(vec![2, 1], vec![2_i64, 0])?)?;
    ///     s.gather(&x, &indices, GatherConfig {
    ///         offset_dims: vec![], collapsed_slice_dims: vec![0],
    ///         start_index_map: vec![0], index_vector_dim: 1,
    ///         slice_sizes: vec![1],
    ///     })
    /// })?;
    /// assert_eq!(result.value()?.as_slice::<f64>()?, &[30.0, 10.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, invalid-index/configuration, or backend error.
    pub fn gather(
        &mut self,
        input: &EagerTensor,
        indices: &EagerTensor,
        config: GatherConfig,
    ) -> Result<EagerTensor> {
        self.ensure_runtime(input)?;
        self.ensure_runtime(indices)?;
        EagerTensor::nary_op_in_session(
            &[input, indices],
            StdTensorOp::Gather(config),
            &mut *self.engine,
        )
    }

    /// Concatenate eager tensors along one axis in this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let result = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let a = s.constant_from(Tensor::from_vec_col_major(vec![1], vec![1.0_f64])?)?;
    ///     let b = s.constant_from(Tensor::from_vec_col_major(vec![1], vec![2.0_f64])?)?;
    ///     s.concatenate(&[&a, &b], 0)
    /// })?;
    /// assert_eq!(result.value()?.as_slice::<f64>()?, &[1.0, 2.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed empty-input, foreign-runtime, invalid-axis/shape, or backend error.
    pub fn concatenate(&mut self, inputs: &[&EagerTensor], axis: usize) -> Result<EagerTensor> {
        for input in inputs {
            self.ensure_runtime(input)?;
        }
        EagerTensor::nary_op_in_session(
            inputs,
            StdTensorOp::Concatenate {
                axis,
                input_count: inputs.len(),
            },
            &mut *self.engine,
        )
    }

    /// Scatter updates into an eager tensor within this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, ScatterConfig, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let result = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let input = s.constant_from(Tensor::from_vec_col_major(vec![4], vec![0.0_f64; 4])?)?;
    ///     let indices = s.constant_from(Tensor::from_vec_col_major(vec![2, 1], vec![1_i64, 3])?)?;
    ///     let updates = s.constant_from(Tensor::from_vec_col_major(vec![2], vec![5.0_f64, 7.0])?)?;
    ///     s.scatter(&input, &indices, &updates, ScatterConfig {
    ///         update_window_dims: vec![],
    ///         inserted_window_dims: vec![0],
    ///         scatter_dims_to_operand_dims: vec![0],
    ///         index_vector_dim: 1,
    ///     })
    /// })?;
    /// assert_eq!(result.value()?.as_slice::<f64>()?, &[0.0, 5.0, 0.0, 7.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, invalid-index/configuration, or backend error.
    pub fn scatter(
        &mut self,
        input: &EagerTensor,
        indices: &EagerTensor,
        updates: &EagerTensor,
        config: ScatterConfig,
    ) -> Result<EagerTensor> {
        self.ensure_runtime(input)?;
        self.ensure_runtime(indices)?;
        self.ensure_runtime(updates)?;
        EagerTensor::nary_op_in_session(
            &[input, indices, updates],
            StdTensorOp::Scatter(config),
            &mut *self.engine,
        )
    }

    /// Extract a diagonal along two axes in this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let diagonal = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let matrix = s.constant_from(Tensor::from_vec_col_major(vec![2, 2], vec![1.0_f64, 2.0, 3.0, 4.0])?)?;
    ///     s.extract_diag(&matrix, 0, 1)
    /// })?;
    /// assert_eq!(diagonal.value()?.as_slice::<f64>()?, &[1.0, 4.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, invalid-axis, or backend error.
    pub fn extract_diag(
        &mut self,
        input: &EagerTensor,
        axis_a: usize,
        axis_b: usize,
    ) -> Result<EagerTensor> {
        self.ensure_runtime(input)?;
        self.run_unary(input, StdTensorOp::ExtractDiag { axis_a, axis_b })
    }

    /// Embed the input along a diagonal in this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let matrix = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let diagonal = s.constant_from(Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0])?)?;
    ///     s.embed_diag(&diagonal, 0, 1)
    /// })?;
    /// assert_eq!(matrix.value()?.as_slice::<f64>()?, &[1.0, 0.0, 0.0, 2.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, invalid-axis, or backend error.
    pub fn embed_diag(
        &mut self,
        input: &EagerTensor,
        axis_a: usize,
        axis_b: usize,
    ) -> Result<EagerTensor> {
        self.ensure_runtime(input)?;
        self.run_unary(input, StdTensorOp::EmbedDiag { axis_a, axis_b })
    }

    /// Reduce an eager tensor over selected axes within this borrowed session.
    /// `None` reduces all axes, while `Some(&[])` retains the input shape.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0])?, ctx.clone())?;
    /// let sum = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| session.reduce_sum(&x, None))?;
    /// assert_eq!(sum.value()?.as_slice::<f64>()?, &[3.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for a foreign runtime, or a typed
    /// validation/backend error for invalid axes or unsupported dtypes.
    pub fn reduce_sum(
        &mut self,
        input: &EagerTensor,
        axes: Option<&[usize]>,
    ) -> Result<EagerTensor> {
        self.ensure_runtime(input)?;
        input.reduce_sum_in_session(axes, &mut *self.engine)
    }

    /// Sum elementwise squares over the selected axes in this borrowed session.
    /// Only `f32` and `f64` are supported. `None` reduces every axis, like the
    /// rest of the reduction family; `Some(&[])` squares each value.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let (sum, all) = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let input = s.constant_from(Tensor::from_vec_col_major([2], vec![3.0_f64, 4.0])?)?;
    ///     Ok::<_, tenferro_ad::Error>((
    ///         s.reduce_sum_squares(&input, Some(&[0]))?,
    ///         s.reduce_sum_squares(&input, None)?,
    ///     ))
    /// })?;
    /// assert_eq!(sum.value()?.as_slice::<f64>()?, &[25.0]);
    /// assert_eq!(all.value()?.as_slice::<f64>()?, &[25.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns typed foreign-runtime, invalid-axis, unsupported-dtype, or backend errors.
    pub fn reduce_sum_squares(
        &mut self,
        input: &EagerTensor,
        axes: Option<&[usize]>,
    ) -> Result<EagerTensor> {
        self.ensure_runtime(input)?;
        let axes = axes.map_or_else(|| (0..input.shape().len()).collect(), <[usize]>::to_vec);
        crate::eager_ops::validate_eager_axes(
            "EagerSession::reduce_sum_squares",
            input.shape().len(),
            &axes,
        )?;
        self.run_unary(input, StdTensorOp::ReduceSumSquares { axes })
    }

    /// Reduce the product of selected axes in this borrowed session.
    /// `None` reduces every axis.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let result = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![2], vec![2.0_f64, 3.0])?)?;
    ///     s.reduce_prod(&x, None)
    /// })?;
    /// assert_eq!(result.value()?.as_slice::<f64>()?, &[6.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, invalid-axis, unsupported-dtype, or backend error.
    pub fn reduce_prod(
        &mut self,
        input: &EagerTensor,
        axes: Option<&[usize]>,
    ) -> Result<EagerTensor> {
        self.ensure_runtime(input)?;
        let axes = axes.map_or_else(|| (0..input.shape().len()).collect(), <[usize]>::to_vec);
        crate::eager_ops::validate_eager_axes(
            "EagerSession::reduce_prod",
            input.shape().len(),
            &axes,
        )?;
        self.run_unary(input, StdTensorOp::ReduceProd { axes })
    }

    /// Reduce the maximum over selected axes in this borrowed session.
    /// `None` reduces every axis.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let result = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![2], vec![2.0_f64, 3.0])?)?;
    ///     s.reduce_max(&x, None)
    /// })?;
    /// assert_eq!(result.value()?.as_slice::<f64>()?, &[3.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, invalid-axis, unsupported-dtype, or backend error.
    pub fn reduce_max(
        &mut self,
        input: &EagerTensor,
        axes: Option<&[usize]>,
    ) -> Result<EagerTensor> {
        self.ensure_runtime(input)?;
        let axes = axes.map_or_else(|| (0..input.shape().len()).collect(), <[usize]>::to_vec);
        crate::eager_ops::validate_eager_axes(
            "EagerSession::reduce_max",
            input.shape().len(),
            &axes,
        )?;
        self.run_unary(input, StdTensorOp::ReduceMax { axes })
    }

    /// Reduce the minimum over selected axes in this borrowed session.
    /// `None` reduces every axis.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let result = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![2], vec![2.0_f64, 3.0])?)?;
    ///     s.reduce_min(&x, None)
    /// })?;
    /// assert_eq!(result.value()?.as_slice::<f64>()?, &[2.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, invalid-axis, unsupported-dtype, or backend error.
    pub fn reduce_min(
        &mut self,
        input: &EagerTensor,
        axes: Option<&[usize]>,
    ) -> Result<EagerTensor> {
        self.ensure_runtime(input)?;
        let axes = axes.map_or_else(|| (0..input.shape().len()).collect(), <[usize]>::to_vec);
        crate::eager_ops::validate_eager_axes(
            "EagerSession::reduce_min",
            input.shape().len(),
            &axes,
        )?;
        self.run_unary(input, StdTensorOp::ReduceMin { axes })
    }

    /// Duplicate an eager value into an independent tensor within the caller's
    /// borrowed session, preserving its dtype and placement.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![1], vec![2.0_f64])?, ctx.clone())?;
    /// let copy = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| session.duplicate_value(&x))?;
    /// assert_eq!(copy.as_slice::<f64>()?, &[2.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for a foreign runtime, or a typed
    /// runtime/backend error when the retained value cannot be duplicated.
    pub fn duplicate_value(&mut self, input: &EagerTensor) -> Result<Tensor> {
        self.ensure_runtime(input)?;
        input.duplicate_value_in_session(&mut *self.engine)
    }

    /// Import an untracked leaf within this borrowed session.
    ///
    /// `tensor` must already be usable by this runtime's backend: a host
    /// tensor on a CPU runtime, or a tensor already on the device of a CUDA
    /// or WebGPU runtime. No host/device transfer happens here. To import host
    /// data into a device runtime, use [`Self::constant_from_host`], which
    /// uploads first; on a CPU runtime the two are equivalent.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let c = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| {
    ///     session.constant_from(Tensor::from_vec_col_major(vec![1], vec![2.0_f64])?)
    /// })?;
    /// assert_eq!(c.value()?.as_slice::<f64>()?, &[2.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::TensorRuntime`] for a typed backend failure when the value cannot be
    /// registered in the session, or [`Error::RuntimeState`] when the runtime's
    /// value registry is unavailable.
    pub fn constant_from(&mut self, tensor: Tensor) -> Result<EagerTensor> {
        EagerTensor::new_leaf_in_session(Arc::clone(self.runtime), tensor, false, &mut *self.engine)
    }

    /// Upload a host tensor and import it as an untracked leaf in this session.
    ///
    /// Unlike [`Self::constant_from`], this explicitly crosses the host/device
    /// boundary: the host `tensor` is uploaded to this runtime's backend
    /// (a host copy on a CPU runtime) and the uploaded value becomes the leaf.
    /// Use it whenever the source data lives on the host and the runtime may
    /// be a device runtime; use [`Self::constant_from`] for a tensor that is
    /// already resident on the backend.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let c = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     s.constant_from_host(Tensor::from_vec_col_major([1], vec![2.0_f64])?)
    /// })?;
    /// assert_eq!(c.value()?.as_slice::<f64>()?, &[2.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns [`Error::TensorRuntime`] for a typed backend failure, including a host-tensor
    /// upload failure, or [`Error::RuntimeState`] when the runtime's value
    /// registry is unavailable.
    pub fn constant_from_host(&mut self, tensor: Tensor) -> Result<EagerTensor> {
        let uploaded = self
            .engine
            .upload_host_tensor(TensorRead::from_tensor(&tensor))
            .map_err(Error::from)?;
        self.constant_from(uploaded)
    }

    /// Import a trainable leaf within this borrowed session.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let x = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| {
    ///     session.variable_from(Tensor::from_vec_col_major(vec![1], vec![2.0_f64])?)
    /// })?;
    /// assert!(x.tracks_grad());
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::TensorRuntime`] for a typed backend failure when the value cannot be
    /// registered, or [`Error::RuntimeState`] when the runtime's value or
    /// gradient registry is unavailable.
    pub fn variable_from(&mut self, tensor: Tensor) -> Result<EagerTensor> {
        EagerTensor::new_leaf_in_session(Arc::clone(self.runtime), tensor, true, &mut *self.engine)
    }

    /// Add eager tensors with the same broadcast and AD rules as the eager
    /// operation surface, reusing this borrowed execution session.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0])?, ctx.clone())?;
    /// let scalar = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![], vec![3.0_f64])?, ctx.clone())?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| session.add(&x, &scalar))?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[4.0, 5.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for a tensor from another runtime,
    /// or a typed broadcast/backend error for the operands.
    pub fn add(&mut self, lhs: &EagerTensor, rhs: &EagerTensor) -> Result<EagerTensor> {
        self.run_binary("add", lhs, rhs, StdTensorOp::Add)
    }

    /// Subtract eager tensors within this borrowed session.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![1], vec![3.0_f64])?, ctx.clone())?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| session.sub(&x, &x))?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[0.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for a foreign runtime, or a typed
    /// broadcast/backend error for the operands.
    pub fn sub(&mut self, lhs: &EagerTensor, rhs: &EagerTensor) -> Result<EagerTensor> {
        self.run_binary("sub", lhs, rhs, StdTensorOp::Sub)
    }

    /// Multiply eager tensors within this borrowed session.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![1], vec![3.0_f64])?, ctx.clone())?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| session.mul(&x, &x))?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[9.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for a foreign runtime, or a typed
    /// broadcast/backend error for the operands.
    pub fn mul(&mut self, lhs: &EagerTensor, rhs: &EagerTensor) -> Result<EagerTensor> {
        self.run_binary("mul", lhs, rhs, StdTensorOp::Mul)
    }

    /// Divide eager tensors elementwise with broadcast rules.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![1], vec![6.0_f64])?)?;
    ///     let divisor = s.constant_from(Tensor::from_vec_col_major(vec![], vec![2.0_f64])?)?;
    ///     s.div(&x, &divisor)
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[3.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns [`Error::ContextMismatch`] when an input belongs to another eager
    /// runtime, a validation error with
    /// `ValidationError::ShapeMismatch` when the operands cannot broadcast, or
    /// [`Error::TensorRuntime`] for a typed backend failure (including integer division by
    /// zero).
    pub fn div(&mut self, lhs: &EagerTensor, rhs: &EagerTensor) -> Result<EagerTensor> {
        self.run_binary("div", lhs, rhs, StdTensorOp::Div)
    }

    /// Compute the elementwise remainder with broadcast rules.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![1], vec![5.0_f64])?)?;
    ///     let divisor = s.constant_from(Tensor::from_vec_col_major(vec![], vec![2.0_f64])?)?;
    ///     s.rem(&x, &divisor)
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[1.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns [`Error::ContextMismatch`] when an input belongs to another eager
    /// runtime, a validation error with
    /// `ValidationError::ShapeMismatch` when the operands cannot broadcast, or
    /// [`Error::TensorRuntime`] for a typed backend failure (including an integer remainder by
    /// zero).
    pub fn rem(&mut self, lhs: &EagerTensor, rhs: &EagerTensor) -> Result<EagerTensor> {
        self.run_binary("rem", lhs, rhs, StdTensorOp::Rem)
    }

    /// Raise eager tensor elements to broadcast exponents.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![1], vec![2.0_f64])?)?;
    ///     let exponent = s.constant_from(Tensor::from_vec_col_major(vec![], vec![3.0_f64])?)?;
    ///     s.pow(&x, &exponent)
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[8.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns [`Error::ContextMismatch`] when an input belongs to another eager
    /// runtime, a validation error with
    /// `ValidationError::ShapeMismatch` when the operands cannot broadcast, or
    /// [`Error::TensorRuntime`] for a typed backend failure (including a negative integer
    /// exponent).
    pub fn pow(&mut self, lhs: &EagerTensor, rhs: &EagerTensor) -> Result<EagerTensor> {
        self.run_binary("pow", lhs, rhs, StdTensorOp::Pow)
    }

    /// Compute the elementwise maximum under broadcast rules.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![1], vec![2.0_f64])?)?;
    ///     let bound = s.constant_from(Tensor::from_vec_col_major(vec![], vec![3.0_f64])?)?;
    ///     s.maximum(&x, &bound)
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[3.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, broadcast, unsupported-dtype, or backend error.
    pub fn maximum(&mut self, lhs: &EagerTensor, rhs: &EagerTensor) -> Result<EagerTensor> {
        self.run_binary("maximum", lhs, rhs, StdTensorOp::Maximum)
    }

    /// Compute the elementwise minimum under broadcast rules.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![1], vec![2.0_f64])?)?;
    ///     let bound = s.constant_from(Tensor::from_vec_col_major(vec![], vec![3.0_f64])?)?;
    ///     s.minimum(&x, &bound)
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[2.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, broadcast, unsupported-dtype, or backend error.
    pub fn minimum(&mut self, lhs: &EagerTensor, rhs: &EagerTensor) -> Result<EagerTensor> {
        self.run_binary("minimum", lhs, rhs, StdTensorOp::Minimum)
    }

    /// Compare eager tensors elementwise under broadcast rules.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{CompareDir, EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![1], vec![2.0_f64])?)?;
    ///     let bound = s.constant_from(Tensor::from_vec_col_major(vec![], vec![1.0_f64])?)?;
    ///     s.compare(&x, &bound, CompareDir::Gt)
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<bool>()?, &[true]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, broadcast, unsupported-dtype, or backend error.
    pub fn compare(
        &mut self,
        lhs: &EagerTensor,
        rhs: &EagerTensor,
        dir: CompareDir,
    ) -> Result<EagerTensor> {
        self.run_binary("compare", lhs, rhs, StdTensorOp::Compare(dir))
    }

    /// Select eager values elementwise using a broadcast boolean condition.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let condition = s.constant_from(Tensor::from_vec_col_major(vec![2], vec![true, false])?)?;
    ///     let yes = s.constant_from(Tensor::from_vec_col_major(vec![], vec![10.0_f64])?)?;
    ///     let no = s.constant_from(Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0])?)?;
    ///     s.where_select(&condition, &yes, &no)
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[10.0, 2.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, broadcast, unsupported-dtype, or backend error.
    pub fn where_select(
        &mut self,
        condition: &EagerTensor,
        on_true: &EagerTensor,
        on_false: &EagerTensor,
    ) -> Result<EagerTensor> {
        self.run_ternary(
            "where_select",
            condition,
            on_true,
            on_false,
            StdTensorOp::Select,
        )
    }

    /// Alias for [`Self::where_select`] with the same borrowed-session semantics.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let predicate = s.constant_from(Tensor::from_vec_col_major(vec![], vec![true])?)?;
    ///     let yes = s.constant_from(Tensor::from_vec_col_major(vec![], vec![3.0_f64])?)?;
    ///     let no = s.constant_from(Tensor::from_vec_col_major(vec![], vec![4.0_f64])?)?;
    ///     s.select(&predicate, &yes, &no)
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[3.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, broadcast, unsupported-dtype, or backend error.
    pub fn select(
        &mut self,
        condition: &EagerTensor,
        on_true: &EagerTensor,
        on_false: &EagerTensor,
    ) -> Result<EagerTensor> {
        self.where_select(condition, on_true, on_false)
    }

    /// Clamp eager values elementwise between broadcast lower and upper bounds.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![2], vec![-2.0_f64, 5.0])?)?;
    ///     let lo = s.constant_from(Tensor::from_vec_col_major(vec![], vec![-1.0_f64])?)?;
    ///     let hi = s.constant_from(Tensor::from_vec_col_major(vec![], vec![4.0_f64])?)?;
    ///     s.clamp(&x, &lo, &hi)
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[-1.0, 4.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, broadcast, unsupported-dtype, or backend error.
    pub fn clamp(
        &mut self,
        input: &EagerTensor,
        lower: &EagerTensor,
        upper: &EagerTensor,
    ) -> Result<EagerTensor> {
        self.run_ternary("clamp", input, lower, upper, StdTensorOp::Clamp)
    }

    /// Contract eager tensors according to a dot-general dimension mapping.
    ///
    /// The output layout is `[lhs free..., rhs free..., batch...]`: batch axes
    /// come last (see [`DotGeneralConfig`](tenferro_runtime::DotGeneralConfig)).
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{DotGeneralConfig, EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let result = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| {
    ///     let lhs = session.variable_from(Tensor::from_vec_col_major(vec![1, 2], vec![2.0_f64, 3.0])?)?;
    ///     let rhs = session.constant_from(Tensor::from_vec_col_major(vec![2, 1], vec![4.0_f64, 5.0])?)?;
    ///     session.dot_general(&lhs, &rhs, DotGeneralConfig {
    ///         lhs_contracting_dims: [1].as_slice().into(),
    ///         rhs_contracting_dims: [0].as_slice().into(),
    ///         lhs_batch_dims: [].as_slice().into(),
    ///         rhs_batch_dims: [].as_slice().into(),
    ///     })
    /// })?;
    /// assert_eq!(result.value()?.as_slice::<f64>()?, &[23.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for a foreign eager runtime,
    /// a typed validation error for incompatible contraction dimensions,
    /// or the backend's typed execution error.
    pub fn dot_general(
        &mut self,
        lhs: &EagerTensor,
        rhs: &EagerTensor,
        config: DotGeneralConfig,
    ) -> Result<EagerTensor> {
        self.ensure_runtime(lhs)?;
        self.ensure_runtime(rhs)?;
        config
            .validate_dims_with_ranks(lhs.shape().len(), rhs.shape().len())
            .map_err(Error::TensorRuntime)?;
        EagerTensor::nary_op_in_session(
            &[lhs, rhs],
            StdTensorOp::DotGeneral { config },
            &mut *self.engine,
        )
    }

    /// Scale an eager tensor by a real scalar in this borrowed session.
    /// Integer factors are rounded; finite zero maps to `false` for boolean inputs.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let scaled = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0])?)?;
    ///     s.scale_real(&x, 2.0)
    /// })?;
    /// assert_eq!(scaled.value()?.as_slice::<f64>()?, &[2.0, 4.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns a typed foreign-runtime, invalid-factor/dtype, or backend error.
    pub fn scale_real(&mut self, input: &EagerTensor, factor: f64) -> Result<EagerTensor> {
        self.ensure_runtime(input)?;
        let scalar = tenferro_runtime::scale::real_scale_scalar(input.dtype(), factor)?;
        let scalar = self.constant_from_host(scalar)?;
        self.mul(input, &scalar)
    }

    /// Scale a complex eager tensor by a complex scalar in this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use num_complex::Complex64;
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let scaled = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.constant_from(Tensor::from_vec_col_major(vec![1], vec![Complex64::new(1.0, 2.0)])?)?;
    ///     s.scale_complex(&x, Complex64::new(0.0, 1.0))
    /// })?;
    /// assert_eq!(scaled.value()?.as_slice::<Complex64>()?, &[Complex64::new(-2.0, 1.0)]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns [`Error::ContextMismatch`] when an input belongs to another eager
    /// runtime, [`Error::TensorRuntime`] containing
    /// `ValidationError::InvalidArgument` when the input dtype is not complex,
    /// or [`Error::TensorRuntime`] for a typed backend failure.
    pub fn scale_complex(&mut self, input: &EagerTensor, factor: Complex64) -> Result<EagerTensor> {
        self.ensure_runtime(input)?;
        let scalar = tenferro_runtime::scale::complex_scale_scalar(input.dtype(), factor)?;
        let scalar = self.constant_from_host(scalar)?;
        self.mul(input, &scalar)
    }

    /// Multiply two rank-2 eager tensors in this borrowed session.
    ///
    /// # Examples
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let result = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let a = s.constant_from(Tensor::from_vec_col_major(vec![1, 1], vec![2.0_f64])?)?;
    ///     let b = s.constant_from(Tensor::from_vec_col_major(vec![1, 1], vec![3.0_f64])?)?;
    ///     s.matmul(&a, &b)
    /// })?;
    /// assert_eq!(result.value()?.as_slice::<f64>()?, &[6.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    /// # Errors
    /// Returns [`Error::ContextMismatch`] when an input belongs to another eager
    /// runtime, a validation error with
    /// `ValidationError::RankMismatch` or `ValidationError::ShapeMismatch` when
    /// the operands are not rank-2 with matching inner dimensions, a dtype
    /// mismatch between the operands, or [`Error::TensorRuntime`] for a typed backend failure.
    pub fn matmul(&mut self, lhs: &EagerTensor, rhs: &EagerTensor) -> Result<EagerTensor> {
        self.ensure_runtime(lhs)?;
        self.ensure_runtime(rhs)?;
        let lhs_shape = lhs.shape();
        let rhs_shape = rhs.shape();
        if lhs_shape.len() != 2 {
            return Err(tenferro_tensor::Error::rank_mismatch("matmul", 2, lhs_shape.len()).into());
        }
        if rhs_shape.len() != 2 {
            return Err(tenferro_tensor::Error::rank_mismatch("matmul", 2, rhs_shape.len()).into());
        }
        if lhs_shape[1] != rhs_shape[0] {
            return Err(
                tenferro_tensor::Error::shape_mismatch("matmul", lhs_shape, rhs_shape).into(),
            );
        }
        self.dot_general(
            lhs,
            rhs,
            DotGeneralConfig {
                lhs_contracting_dims: [1].as_slice().into(),
                rhs_contracting_dims: [0].as_slice().into(),
                lhs_batch_dims: [].as_slice().into(),
                rhs_batch_dims: [].as_slice().into(),
            },
        )
    }

    /// Contract eagerly with optional conjugation of either operand.
    /// Untracked operands use the backend's conjugating contraction directly;
    /// tracked operands record explicit conjugations for reverse-mode AD.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{DotGeneralConfig, EagerRuntime, Tensor};
    /// let ctx = EagerRuntime::new()?;
    /// let result = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| {
    ///     let lhs = session.constant_from(Tensor::from_vec_col_major(vec![1, 1], vec![2.0_f64])?)?;
    ///     let rhs = session.constant_from(Tensor::from_vec_col_major(vec![1, 1], vec![3.0_f64])?)?;
    ///     session.dot_general_with_conj(&lhs, &rhs, DotGeneralConfig {
    ///         lhs_contracting_dims: [1].as_slice().into(),
    ///         rhs_contracting_dims: [0].as_slice().into(),
    ///         lhs_batch_dims: [].as_slice().into(),
    ///         rhs_batch_dims: [].as_slice().into(),
    ///     }, true, false)
    /// })?;
    /// assert_eq!(result.value()?.as_slice::<f64>()?, &[6.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for a foreign eager runtime,
    /// a typed validation error for invalid dimensions, or a backend error.
    pub fn dot_general_with_conj(
        &mut self,
        lhs: &EagerTensor,
        rhs: &EagerTensor,
        config: DotGeneralConfig,
        lhs_conj: bool,
        rhs_conj: bool,
    ) -> Result<EagerTensor> {
        self.ensure_runtime(lhs)?;
        self.ensure_runtime(rhs)?;
        config
            .validate_dims_with_ranks(lhs.shape().len(), rhs.shape().len())
            .map_err(Error::TensorRuntime)?;
        if !lhs.requires_grad && !rhs.requires_grad {
            let output = crate::eager_exec::exec_dot_general_with_conj_on_tensor_reads_in_session(
                lhs.tensor_read(),
                rhs.tensor_read(),
                &config,
                lhs_conj,
                rhs_conj,
                &mut *self.engine,
            )?;
            return EagerTensor::new_untracked_result(Arc::clone(self.runtime), output);
        }
        let lhs = if lhs_conj {
            self.conj(lhs)?
        } else {
            lhs.clone()
        };
        let rhs = if rhs_conj {
            self.conj(rhs)?
        } else {
            rhs.clone()
        };
        self.dot_general(&lhs, &rhs, config)
    }

    fn run_binary(
        &mut self,
        name: &'static str,
        lhs: &EagerTensor,
        rhs: &EagerTensor,
        op: StdTensorOp,
    ) -> Result<EagerTensor> {
        self.ensure_runtime(lhs)?;
        self.ensure_runtime(rhs)?;
        let (lhs, rhs) = crate::eager_ops::broadcast_binary_in_session(name, lhs, rhs, self)?;
        EagerTensor::nary_op_in_session(&[lhs.as_ref(), rhs.as_ref()], op, &mut *self.engine)
    }

    fn run_ternary(
        &mut self,
        name: &'static str,
        first: &EagerTensor,
        second: &EagerTensor,
        third: &EagerTensor,
        op: StdTensorOp,
    ) -> Result<EagerTensor> {
        self.ensure_runtime(first)?;
        self.ensure_runtime(second)?;
        self.ensure_runtime(third)?;
        let (first, second, third) =
            crate::eager_ops::broadcast_ternary_in_session(name, first, second, third, self)?;
        EagerTensor::nary_op_in_session(&[&first, &second, &third], op, &mut *self.engine)
    }

    /// Apply one standard tensor op in this borrowed session and record it
    /// for AD when needed.
    ///
    /// Extension crates use this when an extension-level eager operation
    /// expands into ordinary `StdTensorOp` nodes instead of a custom extension
    /// primitive: all of them run in this one backend session.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// use tenferro_ops::std_tensor_op::StdTensorOp;
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let x = s.variable_from(Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0])?)?;
    ///     let negated = s.apply_standard_op(StdTensorOp::Neg, &[&x])?;
    ///     s.apply_standard_op(StdTensorOp::Mul, &[&negated, &x])
    /// })?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[-1.0, -4.0]);
    /// assert!(y.tracks_grad());
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::TensorRuntime`] containing
    /// [`tenferro_tensor::ValidationError::InvalidArgument`] for an extension
    /// op, [`Error::ContextMismatch`] for a tensor from another runtime, a
    /// typed input-count error, or the backend's typed execution error.
    pub fn apply_standard_op(
        &mut self,
        op: StdTensorOp,
        inputs: &[&EagerTensor],
    ) -> Result<EagerTensor> {
        if matches!(op, StdTensorOp::Extension(_)) {
            return Err(Error::invalid_argument(
                "EagerSession::apply_standard_op",
                ErrorPhase::Execution,
                "op",
                "Extension ops must be passed to apply_eager",
            ));
        }
        for input in inputs {
            self.ensure_runtime(input)?;
        }
        EagerTensor::nary_op_in_session(inputs, op, &mut *self.engine)
    }

    /// Borrow the backend session this eager session runs on.
    ///
    /// Extension crates use it to run their backend kernels on untracked
    /// values inside the same execution region instead of entering a second
    /// session, which would be rejected as reentry. It grants the same access
    /// as passing the engine directly.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// use tenferro_tensor::TensorRead;
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let x = Tensor::from_vec_col_major(vec![2], vec![1.0_f64, -2.0])?;
    /// let copy = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     s.backend_session()
    ///         .to_contiguous_read(TensorRead::from_tensor(&x))
    ///         .map_err(tenferro_ad::Error::from)
    /// })?;
    /// assert_eq!(copy.as_slice::<f64>()?, &[1.0, -2.0]);
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    pub fn backend_session(&mut self) -> &mut dyn Engine {
        &mut *self.engine
    }

    /// Borrow the engine this eager session runs on.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::EagerRuntime;
    /// use tenferro_cpu::CpuEngine;
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let mut engine = CpuEngine::serial();
    /// let id = ctx.with_eager_session(&mut engine, |s| {
    ///     Ok::<_, tenferro_ad::Error>(s.engine().target().engine_id().clone())
    /// })?;
    /// assert_eq!(id, tenferro_cpu::runtime_engine_id()?);
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    pub fn engine(&mut self) -> &mut dyn ExecutionEngine {
        &mut *self.engine
    }

    fn run_unary(&mut self, input: &EagerTensor, op: StdTensorOp) -> Result<EagerTensor> {
        self.ensure_runtime(input)?;
        EagerTensor::nary_op_in_session(&[input], op, &mut *self.engine)
    }

    pub(crate) fn record_outputs(
        &mut self,
        op: &StdTensorOp,
        outputs: &[&Tensor],
        inputs: &[&EagerTensor],
    ) -> Result<RecordedEagerOutputs> {
        record_eager_outputs_in_session(op, outputs, inputs, &mut *self.engine)
    }

    pub(crate) fn ensure_runtime(&self, input: &EagerTensor) -> Result<()> {
        if !Arc::ptr_eq(self.runtime, &input.ctx) {
            return Err(Error::ContextMismatch {
                lhs: self.runtime.id(),
                rhs: input.ctx_id(),
            });
        }
        Ok(())
    }

    pub(crate) fn runtime(&self) -> &Arc<EagerRuntime> {
        self.runtime
    }

    /// Run `f` on this runtime's extension cache store from inside the
    /// session.
    ///
    /// Operation families use this for their own prepared-plan caches without
    /// reopening the runtime: the eager owner is already locked, and the cache
    /// lock is taken second, as in every extension execution region.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::EagerRuntime;
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let entries = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |session| {
    ///     session.with_extension_caches(|caches| caches.len())
    /// })?;
    /// assert_eq!(entries, 0);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns a runtime-state error when the extension cache lock is
    /// poisoned.
    pub fn with_extension_caches<R>(
        &mut self,
        f: impl FnOnce(&mut tenferro_runtime::ExtensionCacheStore) -> R,
    ) -> Result<R> {
        let result = with_lent_extension_caches(self.engine, |_, caches| f(caches));
        Ok(result)
    }
}

impl fmt::Debug for EagerRuntime {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let mut debug = f.debug_struct("EagerRuntime");
        debug.field("id", &self.id);
        debug.field("runtime_id", &self.runtime.id());
        debug.field("runtime_epoch", &self.runtime.epoch().ok());
        match self.extension_install_lock.try_lock() {
            Ok(_) => {
                debug.field("extension_install_lock", &"<unlocked>");
            }
            Err(_) => {
                debug.field("extension_install_lock", &"<locked>");
            }
        }
        debug.field("semantic_extension_rules", &self.semantic_extension_rules);
        match self.grad_slots.try_lock() {
            Ok(slots) => {
                debug.field("grad_slots_len", &slots.len());
            }
            Err(_) => {
                debug.field("grad_slots_len", &"<locked>");
            }
        }
        match self.value_records.try_lock() {
            Ok(records) => {
                debug.field("value_records_len", &records.len());
            }
            Err(_) => {
                debug.field("value_records_len", &"<locked>");
            }
        }
        match self.ad_transform_cache.stats() {
            Ok(stats) => {
                debug.field("ad_transform_cache_stats", &stats);
            }
            Err(err) => {
                debug.field("ad_transform_cache_stats", &format_args!("{err}"));
            }
        }
        match self.prepared_derivative_cache.try_lock() {
            Ok(cache) => {
                debug.field("prepared_derivative_cache_stats", &cache.stats());
            }
            Err(_) => {
                debug.field("prepared_derivative_cache_stats", &"<locked>");
            }
        }
        debug.finish_non_exhaustive()
    }
}

impl EagerRuntime {
    fn lock_extension_install(&self) -> Result<MutexGuard<'_, ()>> {
        self.extension_install_lock.lock().map_err(|_| {
            Error::runtime_state(
                "eager_extension_install",
                ErrorPhase::Execution,
                "lock poisoned",
            )
        })
    }

    fn lock_prepared_derivative_cache(&self) -> Result<MutexGuard<'_, PreparedDerivativeCache>> {
        self.prepared_derivative_cache.lock().map_err(|_| {
            Error::runtime_state(
                "prepared_derivative_cache",
                ErrorPhase::Execution,
                "lock poisoned",
            )
        })
    }

    fn lock_grad_slots(
        &self,
    ) -> Result<MutexGuard<'_, HashMap<ValueKey<StdTensorOp>, WeakGradSlot>>> {
        self.grad_slots.lock().map_err(|_| {
            Error::runtime_state(
                "eager_gradient_slots",
                ErrorPhase::Execution,
                "lock poisoned",
            )
        })
    }

    fn lock_value_records(
        &self,
    ) -> Result<MutexGuard<'_, HashMap<ValueKey<StdTensorOp>, Weak<EagerTensorRecord>>>> {
        self.value_records.lock().map_err(|_| {
            Error::runtime_state(
                "eager_value_registry",
                ErrorPhase::Execution,
                "lock poisoned",
            )
        })
    }

    fn with_rules_and_cache(
        semantic_extension_rules: SemanticExtensionRuleSet,
        ad_transform_cache: Arc<AdTransformCache>,
    ) -> Result<Self> {
        let runtime = Runtime::builder()
            .build()
            .map_err(|source| runtime_config_error("EagerRuntime::new", source))?;
        Ok(Self {
            id: ContextId::fresh(),
            runtime,
            extension_install_lock: Mutex::new(()),
            semantic_extension_rules,
            grad_slots: Mutex::new(HashMap::new()),
            value_records: Mutex::new(HashMap::new()),
            ad_transform_cache,
            prepared_derivative_cache: Mutex::new(PreparedDerivativeCache::default()),
        })
    }

    /// Create a shared eager execution context.
    ///
    /// The context stores no engine: every operation runs on the engine the
    /// caller passes in.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::EagerRuntime;
    ///
    /// let ctx = EagerRuntime::new()?;
    /// assert_eq!(std::sync::Arc::strong_count(&ctx), 1);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::RuntimeStateSource`] when the prepared-program runtime
    /// cannot be configured, preserving the underlying [`RuntimeConfigError`].
    pub fn new() -> Result<Arc<Self>> {
        Ok(Arc::new(Self::with_rules_and_cache(
            SemanticExtensionRuleSet::default(),
            Arc::new(AdTransformCache::new()),
        )?))
    }

    /// Create a shared eager execution context that uses the semantic
    /// extension rules and AD transform cache of `ad`.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{AdContext, EagerRuntime};
    ///
    /// let ad = AdContext::builder().build().unwrap();
    /// let ctx = EagerRuntime::with_ad_context(&ad)?;
    /// assert_eq!(std::sync::Arc::strong_count(&ctx), 1);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::RuntimeStateSource`] when the prepared-program runtime
    /// cannot be configured.
    pub fn with_ad_context(ad: &AdContext) -> Result<Arc<Self>> {
        Ok(Arc::new(Self::with_rules_and_cache(
            ad.semantic_extension_rules().clone(),
            ad.ad_transform_cache(),
        )?))
    }

    /// Register the kind of `engine` with this runtime if it is not registered
    /// yet, and return its target.
    pub(crate) fn ensure_engine(&self, engine: &mut dyn ExecutionEngine) -> Result<EngineTarget> {
        let target = engine.target();
        let snapshot = self
            .runtime
            .snapshot()
            .map_err(|source| runtime_state_source("EagerRuntime::ensure_engine", source))?;
        if let Some(registered) = snapshot.engine(target.engine_id()) {
            if registered.context_identity() != target.context_identity() {
                return Err(Error::unsupported(
                    "EagerRuntime::ensure_engine",
                    ErrorPhase::Execution,
                    format!(
                        "engine id {:?} is registered for a different engine type",
                        target.engine_id()
                    ),
                ));
            }
            return Ok(target);
        }
        let registration = engine_kind_registration(engine)
            .ok_or_else(|| unregisterable_engine_error(target.engine_id()))?;
        let registration = registration
            .map_err(|source| runtime_config_error("EagerRuntime::ensure_engine", source))?;
        let _install_guard = self.lock_extension_install()?;
        let snapshot = self
            .runtime
            .snapshot()
            .map_err(|source| runtime_state_source("EagerRuntime::ensure_engine", source))?;
        if snapshot.engine(target.engine_id()).is_none() {
            self.runtime
                .reconfigure(|edit| {
                    edit.register_engine(registration)?;
                    Ok(())
                })
                .map_err(|source| runtime_state_source("EagerRuntime::ensure_engine", source))?;
        }
        Ok(target)
    }

    /// Return an opaque identifier for this context.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// assert_ne!(ctx.id(), EagerRuntime::new()?.id());
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    pub fn id(&self) -> ContextId {
        self.id
    }

    /// Disable eager operation recording on the current thread until the guard is dropped.
    ///
    /// This is useful for optimizer updates, metric calculations, and other
    /// eager computations that should not become part of the AD tape.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::requires_grad_in(
    ///     Tensor::from_vec_col_major(vec![1], vec![3.0_f64]).unwrap(),
    ///     ctx.clone(),
    /// )?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let _guard = ctx.no_grad();
    ///     s.mul(&x, &x)
    /// })?;
    /// assert!(!y.tracks_grad());
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    pub fn no_grad(&self) -> EagerNoGradGuard {
        EAGER_NO_GRAD_DEPTH.with(|depth| {
            depth.set(depth.get().saturating_add(1));
        });
        EagerNoGradGuard {
            active: true,
            _not_send: PhantomData,
        }
    }

    /// Keep semantic-trace recording active for untracked intermediates.
    ///
    /// See [`EagerTraceCaptureGuard`] for the full contract and an example.
    pub fn capture_trace(&self) -> EagerTraceCaptureGuard {
        EAGER_CAPTURE_DEPTH.with(|depth| {
            depth.set(depth.get().saturating_add(1));
        });
        EagerTraceCaptureGuard {
            active: true,
            _not_send: PhantomData,
        }
    }

    /// Install or replace one extension module on this eager context's runtime.
    ///
    /// Eager extension wrappers call this as an idempotent "ensure installed"
    /// step. When the exact module instance (same module ID and allocation) is
    /// already installed, this is a read-only no-op that returns the current
    /// runtime epoch without acquiring the install lock or reconfiguring. The
    /// cold or replacement paths keep the transactional install-or-replace
    /// behavior, serialized so parallel first-use of the same extension family
    /// cannot publish over another thread's base snapshot.
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] when runtime
    /// reconfiguration fails or the extension module transaction is invalid.
    pub fn install_extension_module(
        &self,
        module: Arc<dyn ExtensionModule>,
    ) -> Result<RuntimeEpoch> {
        let snapshot = self.runtime.snapshot().map_err(|source| {
            runtime_state_source("EagerRuntime::install_extension_module", source)
        })?;
        if snapshot.has_extension_module_identical(&module) {
            return Ok(snapshot.epoch());
        }
        let _install_guard = self.lock_extension_install()?;
        self.runtime
            .reconfigure(|edit| {
                edit.replace_extension_module(module)?;
                Ok(())
            })
            .map_err(|source| {
                runtime_state_source("EagerRuntime::install_extension_module", source)
            })
    }

    pub(crate) fn ensure_extension_module_for_engine(
        &self,
        module: Arc<dyn ExtensionModule>,
        family_id: &'static str,
        engine_id: &EngineId,
    ) -> Result<RuntimeEpoch> {
        let snapshot = self.runtime.snapshot().map_err(|source| {
            runtime_state_source("EagerRuntime::ensure_extension_module_for_engine", source)
        })?;
        if snapshot.has_extension_module_engine(module.module_id(), family_id, engine_id) {
            return Ok(snapshot.epoch());
        }
        let _install_guard = self.lock_extension_install()?;
        self.runtime
            .reconfigure(|edit| {
                edit.ensure_extension_module_for_engine(module, family_id, engine_id)?;
                Ok(())
            })
            .map_err(|source| {
                runtime_state_source("EagerRuntime::ensure_extension_module_for_engine", source)
            })
    }

    pub(crate) fn runtime(&self) -> &Runtime {
        &self.runtime
    }

    /// The extension target for `engine`, registering its kind first.
    pub(crate) fn eager_extension_target(
        &self,
        engine: &mut dyn ExecutionEngine,
    ) -> Result<EagerExtensionTarget> {
        let target = self.ensure_engine(engine)?;
        let backend_kind = eager_extension_backend_kind(&target).ok_or_else(|| {
            Error::unsupported(
                "EagerRuntime::eager_extension_target",
                ErrorPhase::Execution,
                "this engine kind has no eager extension target",
            )
        })?;
        let target = EagerExtensionTarget {
            engine_id: target.engine_id().clone(),
            backend_kind,
        };
        validate_eager_extension_target(&self.runtime, &target)?;
        Ok(target)
    }

    /// Clear every cache owned by this eager context.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// ctx.clear_caches()?;
    /// assert_eq!(ctx.cache_stats()?.ad_transforms.entries, 0);
    /// assert_eq!(ctx.cache_stats()?.prepared_derivatives.entries, 0);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] when either the
    /// extension cache or AD-transform cache is poisoned.
    pub fn clear_caches(&self) -> Result<()> {
        self.clear_ad_transform_caches()?;
        self.clear_prepared_derivative_cache()?;
        Ok(())
    }

    /// Clear prepared derivative program cache entries.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::EagerRuntime;
    ///
    /// let ctx = EagerRuntime::new()?;
    /// ctx.clear_prepared_derivative_cache()?;
    /// assert_eq!(ctx.cache_stats()?.prepared_derivatives.entries, 0);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] if the prepared
    /// derivative cache lock is poisoned.
    pub fn clear_prepared_derivative_cache(&self) -> Result<()> {
        self.lock_prepared_derivative_cache()?.clear();
        Ok(())
    }

    /// Return eager runtime cache-entry and retained-byte stats.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let stats = ctx.cache_stats()?;
    /// assert_eq!(stats.ad_transforms.entries, 0);
    /// assert_eq!(stats.prepared_derivatives.entries, 0);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] when a cache or
    /// AD-transform cache lock is poisoned.
    pub fn cache_stats(&self) -> Result<EagerRuntimeCacheStats> {
        Ok(EagerRuntimeCacheStats {
            ad_transforms: self.ad_transform_cache.stats()?,
            prepared_derivatives: self.lock_prepared_derivative_cache()?.stats(),
        })
    }

    /// Return the AD transform cache retention limits.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::EagerRuntime;
    ///
    /// let ctx = EagerRuntime::new()?;
    /// assert!(ctx.ad_transform_cache_limits()?.max_entries().get() > 0);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] if the AD-transform
    /// cache lock is poisoned.
    pub fn ad_transform_cache_limits(&self) -> Result<AdTransformCacheLimits> {
        self.ad_transform_cache.limits()
    }

    /// Replace AD transform cache retention limits.
    ///
    /// # Examples
    ///
    /// ```
    /// use std::num::NonZeroUsize;
    /// use tenferro_ad::{AdTransformCacheLimits, EagerRuntime};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let limits = AdTransformCacheLimits::new(NonZeroUsize::new(1).unwrap());
    /// ctx.set_ad_transform_cache_limits(limits)?;
    /// assert_eq!(ctx.ad_transform_cache_limits()?, limits);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] if the AD-transform
    /// cache lock is poisoned while updating limits.
    pub fn set_ad_transform_cache_limits(&self, limits: AdTransformCacheLimits) -> Result<()> {
        self.ad_transform_cache.set_limits(limits)
    }

    /// Clear AD transform cache entries visible through this eager runtime.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::EagerRuntime;
    ///
    /// let ctx = EagerRuntime::new()?;
    /// ctx.clear_ad_transform_caches()?;
    /// assert_eq!(ctx.cache_stats()?.ad_transforms.entries, 0);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] if the AD-transform
    /// cache lock is poisoned while clearing entries.
    pub fn clear_ad_transform_caches(&self) -> Result<()> {
        self.ad_transform_cache.clear()
    }

    /// Return prepared derivative cache retention limits.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::EagerRuntime;
    ///
    /// let ctx = EagerRuntime::new()?;
    /// assert!(ctx.prepared_derivative_cache_limits()?.max_entries().get() > 0);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] if the prepared
    /// derivative cache lock is poisoned.
    pub fn prepared_derivative_cache_limits(&self) -> Result<AdTransformCacheLimits> {
        Ok(self.lock_prepared_derivative_cache()?.limits())
    }

    /// Replace prepared derivative cache retention limits.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use std::num::NonZeroUsize;
    /// use tenferro_ad::{AdTransformCacheLimits, EagerRuntime};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let limits = AdTransformCacheLimits::new(NonZeroUsize::new(1).unwrap());
    /// ctx.set_prepared_derivative_cache_limits(limits)?;
    /// assert_eq!(ctx.prepared_derivative_cache_limits()?, limits);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] if the prepared
    /// derivative cache lock is poisoned.
    pub fn set_prepared_derivative_cache_limits(
        &self,
        limits: AdTransformCacheLimits,
    ) -> Result<()> {
        self.lock_prepared_derivative_cache()?.set_limits(limits);
        Ok(())
    }

    /// Run eager operations on `engine`.
    ///
    /// The callback receives an [`EagerSession`] that pairs this runtime with
    /// the borrowed engine; operations recorded through it are differentiable.
    /// Nothing is entered or locked: the engine is simply borrowed for the
    /// callback.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    /// use tenferro_cpu::CpuEngine;
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let mut engine = CpuEngine::serial();
    /// let x = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![1], vec![1.0_f64])?, ctx.clone())?;
    /// let y = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![1], vec![2.0_f64])?, ctx.clone())?;
    /// let z = ctx.with_eager_session(&mut engine, |s| s.add(&x, &y))?;
    /// assert_eq!(z.value()?.as_slice::<f64>()?, &[3.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::Unsupported`] when the engine kind cannot be registered
    /// with this runtime or its context identity conflicts with the existing
    /// registration, [`Error::RuntimeStateSource`] when the runtime snapshot or
    /// engine registration state is unavailable, and the callback's own error
    /// unchanged.
    pub fn with_eager_session<T, E: From<Error>>(
        self: &Arc<Self>,
        engine: &mut dyn ExecutionEngine,
        f: impl FnOnce(&mut EagerSession<'_>) -> std::result::Result<T, E>,
    ) -> std::result::Result<T, E> {
        self.ensure_engine(engine).map_err(E::from)?;
        f(&mut EagerSession {
            runtime: self,
            engine,
        })
    }

    /// Run a compiled program on `engine` through this runtime.
    pub(crate) fn run_program(
        &self,
        engine: &mut dyn ExecutionEngine,
        program: &CompiledGraph,
        inputs: &[TensorRead<'_>],
    ) -> Result<Vec<Tensor>> {
        let target = self.ensure_engine(engine)?;
        let prepared = self
            .runtime
            .prepare(program, inputs, std::slice::from_ref(&target))?;
        run_prepared_on(&self.runtime, engine, &prepared, inputs)
    }

    pub(crate) fn exec_extension_outputs_read(
        &self,
        engine: &mut dyn ExecutionEngine,
        op: &Arc<dyn tenferro_ops::ext_op::ExtensionOp>,
        inputs: &[TensorRead<'_>],
    ) -> Result<Vec<Tensor>> {
        self.ensure_engine(engine)?;
        profile_eager_op_section("exec_extension_outputs_read.exec_op", || {
            exec_extension_op_on_tensor_reads(op, inputs, engine, &self.runtime)
        })
    }

    #[cfg(test)]
    pub(crate) fn exec_standard_graph_outputs(
        &self,
        engine: &mut dyn ExecutionEngine,
        graph: &Graph<StdTensorOp>,
        initial_data: HashMap<ValueKey<StdTensorOp>, Tensor>,
    ) -> Result<EagerGraphExecution> {
        let mut all_values = initial_data;

        profile_eager_op_section("exec_graph.run", || {
            let exec: &mut dyn Engine = engine;
            (|| -> Result<()> {
                for op_node in graph.operations() {
                    let outputs = {
                        let input_values = op_node
                            .inputs
                            .iter()
                            .map(|input| {
                                let key = match input {
                                    ValueRef::Local(local_id) => &graph.values()[*local_id].key,
                                    ValueRef::External(key) => key,
                                };
                                all_values.get(key).ok_or_else(|| {
                                    Error::Internal(format!(
                                        "standard graph eager execution missing value for {key:?}"
                                    ))
                                })
                            })
                            .collect::<Result<Vec<_>>>()?;
                        let input_reads = input_values
                            .iter()
                            .map(|value| TensorRead::from_tensor(value))
                            .collect::<Vec<_>>();
                        exec_standard_op_on_tensor_reads_in_session(
                            &op_node.operation,
                            &input_reads,
                            exec,
                        )?
                    };

                    if outputs.len() != op_node.outputs.len() {
                        return Err(Error::Internal(format!(
                            "standard graph eager execution expected {} outputs for {:?}, got {}",
                            op_node.outputs.len(),
                            op_node.operation,
                            outputs.len()
                        )));
                    }

                    for (output_id, output) in op_node.outputs.iter().zip(outputs) {
                        let key = graph.values()[*output_id].key.clone();
                        all_values.insert(key, output);
                    }
                }
                Ok(())
            })()
        })?;

        let outputs = graph
            .outputs()
            .iter()
            .map(|&output_id| {
                let key = &graph.values()[output_id].key;
                all_values
                    .get(key)
                    .ok_or_else(|| {
                        Error::Internal(format!(
                            "standard graph eager execution missing graph output {key:?}"
                        ))
                    })?
                    .duplicate()
                    .map_err(Error::from)
            })
            .collect::<Result<Vec<_>>>()?;

        Ok(EagerGraphExecution { outputs })
    }

    pub(crate) fn try_register_grad_slot(
        &self,
        key: &ValueKey<StdTensorOp>,
        slot: &GradSlot,
    ) -> Result<()> {
        insert_pruning_dead(
            &mut *self.lock_grad_slots()?,
            key.clone(),
            Arc::downgrade(slot),
        );
        Ok(())
    }

    pub(crate) fn try_register_value_record(
        &self,
        key: &ValueKey<StdTensorOp>,
        record: &Arc<EagerTensorRecord>,
    ) -> Result<()> {
        insert_pruning_dead(
            &mut *self.lock_value_records()?,
            key.clone(),
            Arc::downgrade(record),
        );
        Ok(())
    }

    pub(crate) fn value_record(
        &self,
        key: &ValueKey<StdTensorOp>,
    ) -> Result<Option<Arc<EagerTensorRecord>>> {
        let mut records = self.lock_value_records()?;
        let Some(record) = records.get(key).cloned() else {
            return Ok(None);
        };
        match record.upgrade() {
            Some(record) => Ok(Some(record)),
            None => {
                records.remove(key);
                Ok(None)
            }
        }
    }

    /// Clear all live gradient slots tracked by this context.
    ///
    /// This resets the stored gradients to `None` without unregistering the
    /// tensors, so future `backward()` calls can accumulate again.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::requires_grad_in(Tensor::from_vec_col_major(vec![3], vec![1.0_f64, 2.0, 3.0]).unwrap(), ctx.clone()).unwrap();
    /// let y = EagerTensor::requires_grad_in(Tensor::from_vec_col_major(vec![3], vec![4.0_f64, 5.0, 6.0]).unwrap(), ctx.clone()).unwrap();
    /// let loss = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let product = s.mul(&x, &y)?;
    ///     s.reduce_sum(&product, Some(&[0]))
    /// })?;
    /// let _ = loss.backward(&mut tenferro_cpu::CpuEngine::serial()).unwrap();
    ///
    /// ctx.clear_grads()?;
    ///
    /// assert!(x.grad()?.is_none());
    /// assert!(y.grad()?.is_none());
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] if a gradient-slot
    /// lock is poisoned while clearing live gradients.
    pub fn clear_grads(&self) -> Result<()> {
        let live_slots = {
            let mut live_slots = Vec::new();
            self.lock_grad_slots()?.retain(|_, slot| {
                if let Some(slot) = slot.upgrade() {
                    live_slots.push(slot);
                    true
                } else {
                    false
                }
            });
            live_slots
        };

        let mut poisoned_slot = false;
        for slot in live_slots {
            match slot.lock() {
                Ok(mut current) => {
                    *current = None;
                }
                Err(_) => {
                    poisoned_slot = true;
                }
            }
        }
        if poisoned_slot {
            return Err(Error::runtime_state(
                "eager_gradient_slot",
                ErrorPhase::Execution,
                "lock poisoned",
            ));
        }
        Ok(())
    }

    /// Import a concrete tensor into this context as an untracked constant.
    ///
    /// The returned tensor does not participate in gradient tracking.
    /// Use this for fixed masks, quadrature weights, physical constants,
    /// and other data that should not receive gradients. Like
    /// [`EagerSession::constant_from`], it performs no host/device transfer;
    /// use [`EagerSession::constant_from_host`] to upload host data into a
    /// device runtime.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let c = ctx.constant_from(Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0]).unwrap())?;
    /// let x = EagerTensor::requires_grad_in(Tensor::from_vec_col_major(vec![2], vec![3.0_f64, 4.0]).unwrap(), ctx.clone())?;
    /// let z = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| s.add(&x, &c))?;
    ///
    /// assert_eq!(z.value()?.as_slice::<f64>().unwrap(), &[4.0, 6.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] when metadata cannot
    /// be registered or the backend lock is poisoned.
    pub fn constant_from(self: &Arc<Self>, tensor: Tensor) -> Result<EagerTensor> {
        EagerTensor::new_leaf(Arc::clone(self), tensor, false)
    }

    /// Import a concrete tensor into this context as a trainable variable.
    ///
    /// The returned tensor participates in gradient tracking; its gradient
    /// slot is registered in this context.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let p = ctx.variable_from(Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0]).unwrap())?;
    /// let loss = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let y = s.exp(&p)?;
    ///     s.reduce_sum(&y, Some(&[0]))
    /// })?;
    /// let _ = loss.backward(&mut tenferro_cpu::CpuEngine::serial()).unwrap();
    ///
    /// let grad = p.grad().unwrap().unwrap();
    /// assert_eq!(grad.shape(), &[2]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] when gradient metadata
    /// or the eager backend state cannot be registered.
    pub fn variable_from(self: &Arc<Self>, tensor: Tensor) -> Result<EagerTensor> {
        EagerTensor::new_leaf(Arc::clone(self), tensor, true)
    }

    /// Gradient of a scalar eager output with respect to an eager tensor.
    ///
    /// Functional eager gradients return ordinary eager tensors and do not
    /// write into `grad()` slots. The returned tensor keeps a trace when the
    /// derivative computation depends on tracked eager values.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::requires_grad_in(
    ///     Tensor::from_vec_col_major(vec![], vec![3.0_f64]).unwrap(),
    ///     ctx.clone(),
    /// )?;
    /// let loss = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| s.mul(&x, &x))?;
    /// let dx = ctx.grad(&mut tenferro_cpu::CpuEngine::serial(), &loss, &x)?;
    /// assert_eq!(dx.value()?.as_slice::<f64>().unwrap(), &[6.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::NonScalarGrad`] for a non-scalar
    /// output, [`Error::ContextMismatch`] for tensors from another runtime,
    /// [`Error::UnsupportedAdRule`] when an AD rule is unavailable, or a typed
    /// validation/backend error from eager execution. An inactive `wrt` returns
    /// [`Error::Validation`] with `argument: "wrt"`; use
    /// [`grad_optional`](Self::grad_optional) to observe that state.
    pub fn grad(
        self: &Arc<Self>,
        engine: &mut dyn ExecutionEngine,
        output: &EagerTensor,
        wrt: &EagerTensor,
    ) -> Result<EagerTensor> {
        self.grad_optional(engine, output, wrt)?
            .ok_or_else(|| crate::traced::inactive_wrt_error("grad", &wrt.key))
    }

    /// Gradient that returns `None` when `wrt` is inactive.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::requires_grad_in(
    ///     Tensor::from_vec_col_major(vec![], vec![3.0_f64]).unwrap(),
    ///     ctx.clone(),
    /// )?;
    /// let y = EagerTensor::requires_grad_in(
    ///     Tensor::from_vec_col_major(vec![], vec![4.0_f64]).unwrap(),
    ///     ctx.clone(),
    /// )?;
    /// let loss = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| s.mul(&y, &y))?;
    /// assert!(ctx.grad_optional(&mut tenferro_cpu::CpuEngine::serial(), &loss, &x)?.is_none());
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::NonScalarGrad`] for a non-scalar
    /// output, [`Error::ContextMismatch`] for a foreign runtime, or a typed
    /// validation/backend/runtime-state error from eager execution.
    pub fn grad_optional(
        self: &Arc<Self>,
        engine: &mut dyn ExecutionEngine,
        output: &EagerTensor,
        wrt: &EagerTensor,
    ) -> Result<Option<EagerTensor>> {
        if !output.shape().is_empty() {
            return Err(Error::NonScalarGrad {
                shape: output.shape().to_vec(),
            });
        }

        let value = output.to_tensor(&mut *engine)?;
        let seed = one_like_tensor(&value, &mut *engine)?;
        let seed = EagerTensor::new_result(Arc::clone(self), eager_val_key(), seed, false, None)?;
        self.vjp_optional(engine, output, wrt, &seed)
    }

    /// Reverse-mode vector-Jacobian product for eager tensors.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::requires_grad_in(
    ///     Tensor::from_vec_col_major(vec![2], vec![2.0_f64, 3.0]).unwrap(),
    ///     ctx.clone(),
    /// )?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| s.mul(&x, &x))?;
    /// let seed = EagerTensor::from_tensor_in(
    ///     Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 1.0]).unwrap(),
    ///     ctx.clone(),
    /// )?;
    /// let dx = ctx.vjp(&mut tenferro_cpu::CpuEngine::serial(), &y, &x, &seed)?;
    /// assert_eq!(dx.value()?.as_slice::<f64>().unwrap(), &[4.0, 6.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for tensors from different eager
    /// runtimes, [`Error::Validation`] when the cotangent shape or dtype does
    /// not match the output, [`Error::UnsupportedAdRule`] when a rule is not
    /// registered, or a typed backend/runtime-state error. An inactive `wrt`
    /// returns [`Error::Validation`] with `argument: "wrt"`; use
    /// [`vjp_optional`](Self::vjp_optional) to observe that state.
    pub fn vjp(
        self: &Arc<Self>,
        engine: &mut dyn ExecutionEngine,
        output: &EagerTensor,
        wrt: &EagerTensor,
        cotangent: &EagerTensor,
    ) -> Result<EagerTensor> {
        self.vjp_optional(engine, output, wrt, cotangent)?
            .ok_or_else(|| crate::traced::inactive_wrt_error("vjp", &wrt.key))
    }

    /// Reverse-mode vector-Jacobian product that returns `None` for inactive inputs.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::requires_grad_in(
    ///     Tensor::from_vec_col_major(vec![1], vec![2.0_f64]).unwrap(),
    ///     ctx.clone(),
    /// )?;
    /// let y = EagerTensor::requires_grad_in(
    ///     Tensor::from_vec_col_major(vec![1], vec![4.0_f64]).unwrap(),
    ///     ctx.clone(),
    /// )?;
    /// let seed = EagerTensor::from_tensor_in(
    ///     Tensor::from_vec_col_major(vec![1], vec![1.0_f64]).unwrap(),
    ///     ctx.clone(),
    /// )?;
    /// let loss = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| s.mul(&y, &y))?;
    /// assert!(ctx.vjp_optional(&mut tenferro_cpu::CpuEngine::serial(), &loss, &x, &seed)?.is_none());
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for tensors from different eager
    /// runtimes, [`Error::Validation`] when the cotangent shape or dtype does
    /// not match the output, [`Error::UnsupportedAdRule`] when a rule is not
    /// registered, or a typed backend/runtime-state error.
    pub fn vjp_optional(
        self: &Arc<Self>,
        engine: &mut dyn ExecutionEngine,
        output: &EagerTensor,
        wrt: &EagerTensor,
        cotangent: &EagerTensor,
    ) -> Result<Option<EagerTensor>> {
        validate_same_runtime(self, output, "vjp output")?;
        validate_same_runtime(self, wrt, "vjp wrt")?;
        validate_same_runtime(self, cotangent, "vjp cotangent")?;
        validate_seed_tensor("vjp", output, cotangent)?;
        Ok(
            semantic_eager_vjp_many(self, engine, output, &[wrt], cotangent)?
                .pop()
                .flatten(),
        )
    }

    /// Forward-mode Jacobian-vector product for eager tensors.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::requires_grad_in(
    ///     Tensor::from_vec_col_major(vec![1], vec![3.0_f64]).unwrap(),
    ///     ctx.clone(),
    /// )?;
    /// let tangent = EagerTensor::from_tensor_in(
    ///     Tensor::from_vec_col_major(vec![1], vec![1.0_f64]).unwrap(),
    ///     ctx.clone(),
    /// )?;
    /// let y = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| s.mul(&x, &x))?;
    /// let dy = ctx.jvp(&mut tenferro_cpu::CpuEngine::serial(), &y, &x, &tangent)?;
    /// assert_eq!(dy.value()?.as_slice::<f64>().unwrap(), &[6.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for tensors from different eager
    /// runtimes, [`Error::Validation`] when the tangent shape or dtype does not
    /// match `wrt`, [`Error::UnsupportedAdRule`] when a rule is unavailable, or
    /// a typed backend/runtime-state error. An inactive `wrt` returns
    /// [`Error::Validation`] with `argument: "wrt"`; use
    /// [`jvp_optional`](Self::jvp_optional) to observe that state.
    pub fn jvp(
        self: &Arc<Self>,
        engine: &mut dyn ExecutionEngine,
        output: &EagerTensor,
        wrt: &EagerTensor,
        tangent: &EagerTensor,
    ) -> Result<EagerTensor> {
        self.jvp_optional(engine, output, wrt, tangent)?
            .ok_or_else(|| crate::traced::inactive_wrt_error("jvp", &wrt.key))
    }

    /// Forward-mode Jacobian-vector product that returns `None` for inactive outputs.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::requires_grad_in(
    ///     Tensor::from_vec_col_major(vec![1], vec![2.0_f64]).unwrap(),
    ///     ctx.clone(),
    /// )?;
    /// let y = EagerTensor::requires_grad_in(
    ///     Tensor::from_vec_col_major(vec![1], vec![4.0_f64]).unwrap(),
    ///     ctx.clone(),
    /// )?;
    /// let tangent = EagerTensor::from_tensor_in(
    ///     Tensor::from_vec_col_major(vec![1], vec![1.0_f64]).unwrap(),
    ///     ctx.clone(),
    /// )?;
    /// let loss = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| s.mul(&y, &y))?;
    /// assert!(ctx.jvp_optional(&mut tenferro_cpu::CpuEngine::serial(), &loss, &x, &tangent)?.is_none());
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] for tensors from different eager
    /// runtimes, [`Error::Validation`] when the tangent shape or dtype does not
    /// match `wrt`, [`Error::UnsupportedAdRule`] when a rule is unavailable, or
    /// a typed backend/runtime-state error.
    pub fn jvp_optional(
        self: &Arc<Self>,
        engine: &mut dyn ExecutionEngine,
        output: &EagerTensor,
        wrt: &EagerTensor,
        tangent: &EagerTensor,
    ) -> Result<Option<EagerTensor>> {
        validate_same_runtime(self, output, "jvp output")?;
        validate_same_runtime(self, wrt, "jvp wrt")?;
        validate_same_runtime(self, tangent, "jvp tangent")?;
        validate_seed_tensor("jvp", wrt, tangent)?;
        // Unification 7: semantic path is the only JVP path.
        match semantic_eager_jvp_optional(self, engine, output, wrt, tangent)? {
            Some(result) => Ok(result),
            None => Ok(None),
        }
    }

    fn store_grads(
        &self,
        cotangents: &HashMap<ValueKey<StdTensorOp>, Tensor>,
        session: &mut dyn Engine,
    ) -> Result<()> {
        let mut updates = Vec::new();

        {
            let mut slots = self.lock_grad_slots()?;
            slots.retain(|key, slot| {
                let Some(slot) = slot.upgrade() else {
                    return false;
                };

                if let Some(incoming) = cotangents.get(key) {
                    updates.push((slot, incoming));
                }

                true
            });
        }

        for (slot, incoming) in updates {
            let mut current = slot.lock().map_err(|_| {
                Error::runtime_state(
                    "eager_gradient_slot",
                    ErrorPhase::Execution,
                    "lock poisoned",
                )
            })?;
            let next = match current.as_ref() {
                Some(existing) => {
                    let existing_read = existing.tensor_read("EagerRuntime::store_grads")?;
                    let incoming_read = TensorRead::from_tensor(incoming);
                    let tensor = session
                        .add_read(existing_read, incoming_read)
                        .map_err(Error::from)?;
                    AdValueRecord::from_tensor(tensor, "EagerRuntime::store_grads")?
                }
                None => {
                    let duplicate = session
                        .to_contiguous_read(TensorRead::from_tensor(incoming))
                        .map_err(Error::from)?;
                    AdValueRecord::from_tensor(duplicate, "EagerRuntime::store_grads")?
                }
            };
            *current = Some(next);
        }

        Ok(())
    }
}

#[derive(Clone, Debug, PartialEq, Eq, Hash)]
struct PreparedDerivativeCacheKey {
    /// Engines with the same target share prepared derivative programs.
    target: EngineTarget,
    semantic_fingerprint: SemanticFingerprint,
    runtime_epoch: RuntimeEpoch,
    active_inputs: Box<[bool]>,
    input_metadata: Box<[ProgramValueMetadata]>,
}

/// Cached prepared derivative: program + index metadata.
#[derive(Debug)]
struct PreparedDerivative {
    program: Arc<CompiledGraph>,
    execution_program: Arc<CompiledGraph>,
    saved_input_indices: Vec<usize>,
    prepared: Arc<Prepared>,
    seed_input_index: usize,
    derivative_output_indices: Box<[Option<usize>]>,
}

#[derive(Debug)]
struct PreparedDerivativeCache {
    limits: AdTransformCacheLimits,
    entries: LruCache<PreparedDerivativeCacheKey, PreparedDerivativeCacheEntry>,
    stats: CacheStats,
}

impl PreparedDerivativeCache {
    fn limits(&self) -> AdTransformCacheLimits {
        self.limits
    }

    fn set_limits(&mut self, limits: AdTransformCacheLimits) {
        self.limits = limits;
        self.evict_to_limits();
    }

    fn clear(&mut self) {
        let clears = self.stats.clears.saturating_add(1);
        self.entries.clear();
        self.stats = CacheStats {
            clears,
            ..CacheStats::empty()
        };
    }

    fn stats(&self) -> CacheStats {
        self.stats
    }

    fn get(&mut self, key: &PreparedDerivativeCacheKey) -> Option<Arc<PreparedDerivative>> {
        match self.entries.get(key) {
            Some(entry) => {
                self.stats.hits = self.stats.hits.saturating_add(1);
                Some(Arc::clone(&entry.value))
            }
            None => {
                self.stats.misses = self.stats.misses.saturating_add(1);
                None
            }
        }
    }

    fn insert(&mut self, key: PreparedDerivativeCacheKey, value: Arc<PreparedDerivative>) {
        let retained_bytes = prepared_derivative_cache_entry_retained_bytes(&key, value.as_ref());
        let entry = PreparedDerivativeCacheEntry {
            value,
            retained_bytes,
        };
        self.stats.retained_bytes = self.stats.retained_bytes.saturating_add(retained_bytes);
        if let Some((_old_key, old_entry)) = self.entries.push(key, entry) {
            self.stats.retained_bytes = self
                .stats
                .retained_bytes
                .saturating_sub(old_entry.retained_bytes);
        }
        self.stats.entries = self.entries.len();
        self.evict_to_limits();
    }

    fn evict_to_limits(&mut self) {
        while self.entries.len() > self.limits.max_entries().get()
            || self
                .limits
                .max_retained_bytes()
                .is_some_and(|limit| self.stats.retained_bytes > limit.get())
        {
            let Some((_key, entry)) = self.entries.pop_lru() else {
                break;
            };
            self.stats.retained_bytes = self
                .stats
                .retained_bytes
                .saturating_sub(entry.retained_bytes);
            self.stats.evictions = self.stats.evictions.saturating_add(1);
        }
        self.stats.entries = self.entries.len();
    }
}

impl Default for PreparedDerivativeCache {
    fn default() -> Self {
        Self {
            limits: AdTransformCacheLimits::default(),
            entries: LruCache::unbounded(),
            stats: CacheStats::empty(),
        }
    }
}

#[derive(Debug)]
struct PreparedDerivativeCacheEntry {
    value: Arc<PreparedDerivative>,
    retained_bytes: usize,
}

fn prepared_derivative_cache_entry_retained_bytes(
    key: &PreparedDerivativeCacheKey,
    value: &PreparedDerivative,
) -> usize {
    size_of::<PreparedDerivativeCacheKey>()
        .saturating_add(size_of_val(key.active_inputs.as_ref()))
        .saturating_add(size_of_val(value.derivative_output_indices.as_ref()))
        .saturating_add(size_of_val(value.saved_input_indices.as_slice()))
        .saturating_add(compiled_graph_retained_bytes(
            value.execution_program.as_ref(),
        ))
        .saturating_add(
            key.input_metadata
                .len()
                .saturating_mul(size_of::<ProgramValueMetadata>()),
        )
        .saturating_add(size_of::<PreparedDerivative>())
        .saturating_add(compiled_graph_retained_bytes(value.program.as_ref()))
        .saturating_add(prepared_compiled_graph_retained_bytes(
            value.prepared.as_ref(),
            value.program.as_ref(),
        ))
}

fn prepared_compiled_graph_retained_bytes(
    prepared: &Prepared,
    derivative_program: &CompiledGraph,
) -> usize {
    size_of_val(prepared).saturating_add(compiled_graph_retained_bytes(derivative_program))
}

fn compiled_graph_retained_bytes(program: &CompiledGraph) -> usize {
    size_of::<CompiledGraph>()
        .saturating_add(size_of_val(program.input_keys()))
        .saturating_add(program.bindings().len().saturating_mul(size_of::<usize>()))
        .saturating_add(semantic_program_retained_bytes(program.program()))
}

fn semantic_program_retained_bytes(program: &SemanticProgram) -> usize {
    size_of::<SemanticProgram>()
        .saturating_add(size_of_val(program.inputs()))
        .saturating_add(size_of_val(program.outputs()))
        .saturating_add(
            program
                .operations()
                .len()
                .saturating_mul(size_of::<usize>()),
        )
        .saturating_add(
            program
                .shape_guards()
                .len()
                .saturating_mul(size_of::<usize>()),
        )
}

fn semantic_eager_vjp_many(
    ctx: &Arc<EagerRuntime>,
    engine: &mut dyn ExecutionEngine,
    output: &EagerTensor,
    wrts: &[&EagerTensor],
    cotangent: &EagerTensor,
) -> Result<Vec<Option<EagerTensor>>> {
    if !eager_semantic_vjp_enabled() || wrts.is_empty() {
        return Ok(vec![None; wrts.len()]);
    }
    let Some(raw_output_trace) = output.semantic_trace.as_ref() else {
        return Ok(vec![None; wrts.len()]);
    };
    if !wrts.iter().any(|wrt| {
        wrt.semantic_trace
            .as_ref()
            .and_then(TracedTensor::input_key)
            .is_some_and(|key| raw_output_trace.has_attached_input_key(&key))
    }) {
        return Ok(vec![None; wrts.len()]);
    }

    // First AD request on this output: run the deferred graph analysis over
    // the whole raw carrier chain once (metadata registration + constraint
    // scopes), so `compile_ad_source` sees the same analyzed graph the eager
    // forward used to append.
    let output_trace = analyze_deferred_semantic_trace(raw_output_trace)?;
    let saved = output
        .trace
        .as_ref()
        .map(EagerTrace::collect)
        .unwrap_or_default();
    let mut source_outputs = vec![&output_trace];
    source_outputs.extend(saved.iter().map(|value| &value.trace));

    // Residual roots keep the semantic producers available to differentiation;
    // only the execution-only derivative replaces their numerical values.
    let mut compiler = GraphCompiler::new();
    let source =
        tenferro_runtime::ad_support::compile_ad_source_many(&mut compiler, &source_outputs)?;
    if source.output_count() != source_outputs.len()
        || source.input_keys().len() != source.input_count()
        || source.bindings().len() != source.input_count()
    {
        return Ok(vec![None; wrts.len()]);
    }
    let wrt_input_indices = wrts
        .iter()
        .map(|wrt| {
            let key = wrt.semantic_trace.as_ref()?.input_key()?;
            source.input_key_index(&key)
        })
        .collect::<Vec<_>>();
    let mut active_inputs = vec![false; source.input_count()];
    for &index in wrt_input_indices.iter().flatten() {
        active_inputs[index] = true;
    }
    if !active_inputs.iter().any(|&active| active) {
        return Ok(vec![None; wrts.len()]);
    }

    // One transform and execution for the complete active set shares primal
    // work and cotangents between leaves instead of replaying them per target.
    // S2: check prepared-derivative cache before AD transform + compile_frozen.
    let target = ctx.ensure_engine(engine)?;
    let cache_key = PreparedDerivativeCacheKey {
        target: target.clone(),
        semantic_fingerprint: source.program().semantic_fingerprint(),
        runtime_epoch: ctx.runtime.epoch().map_err(|source| {
            Error::runtime_state_source("semantic_eager_vjp", ErrorPhase::Execution, source)
        })?,
        active_inputs: active_inputs.clone().into_boxed_slice(),
        input_metadata: source.frozen_program().input_metadata_with_bound_shapes(),
    };
    let prepared = { ctx.lock_prepared_derivative_cache()?.get(&cache_key) };
    let (
        seed_input_index,
        derivative_output_indices,
        derivative_program,
        execution_program,
        saved_input_indices,
        prepared_runtime,
    ) = if let Some(prepared) = prepared {
        (
            prepared.seed_input_index,
            prepared.derivative_output_indices.clone(),
            Arc::clone(&prepared.program),
            Arc::clone(&prepared.execution_program),
            prepared.saved_input_indices.clone(),
            Some(Arc::clone(&prepared.prepared)),
        )
    } else {
        let mut active_outputs = vec![false; source.output_count()];
        active_outputs[0] = true;
        let ad = AdContext::with_rules_and_transform_cache(
            ctx.semantic_extension_rules.clone(),
            Arc::clone(&ctx.ad_transform_cache),
        );
        let derivative = ad
            .vjp_program(source.frozen_program(), &active_inputs, &active_outputs)
            .map_err(|source| {
                Error::runtime_state_source("semantic_eager_vjp", ErrorPhase::GraphBuild, source)
            })?;
        let seed_input_index = derivative
            .derivative_input_indices()
            .first()
            .copied()
            .flatten();
        let Some(seed_input_index) = seed_input_index else {
            return Ok(vec![None; wrts.len()]);
        };
        let derivative_output_indices = derivative
            .derivative_output_indices()
            .to_vec()
            .into_boxed_slice();
        let program = Arc::new(compiler.compile_frozen_program(derivative.frozen())?);
        let (execution_program, saved_input_indices) = if saved.is_empty() {
            (Arc::clone(&program), Vec::new())
        } else {
            let (execution, indices) = crate::semantic_transform::semantic_vjp_with_saved_outputs(
                source.frozen_program(),
                &active_inputs,
                &active_outputs,
                &ctx.semantic_extension_rules,
                &(1..source.output_count()).collect::<Vec<_>>(),
            )
            .map_err(|source| {
                Error::runtime_state_source("semantic_eager_vjp", ErrorPhase::GraphBuild, source)
            })?;
            (
                Arc::new(compiler.compile_frozen_program(execution.frozen())?),
                indices,
            )
        };
        (
            seed_input_index,
            derivative_output_indices,
            program,
            execution_program,
            saved_input_indices,
            None,
        )
    };

    let cotangent_tensor = Arc::new(RetainedValue::from_tensor(
        cotangent.to_tensor(&mut *engine)?,
    ));
    let input_count = execution_program.input_count();
    let mut owned_inputs: Vec<Option<Tensor>> = (0..input_count).map(|_| None).collect();
    // Residuals, primal bindings and the seed are staged on the caller's engine.
    (|| -> Result<()> {
        let session: &mut dyn Engine = &mut *engine;
        for (value, &index) in saved.iter().zip(&saved_input_indices) {
            let read = value.value.tensor_read("eager residual")?;
            let Some(slot) = owned_inputs.get_mut(index) else {
                return Err(Error::Internal(format!(
                    "semantic eager VJP residual index {index} is outside {input_count} inputs"
                )));
            };
            *slot = Some(session.to_contiguous_read(read)?);
        }
        for (source_input_index, (_, tensor)) in source.bindings().iter().enumerate() {
            let Some(slot) = owned_inputs.get_mut(source_input_index) else {
                return Err(Error::Internal(format!(
                    "semantic eager VJP derivative program has no primal input slot {source_input_index}"
                )));
            };
            *slot = Some(copy_value_in_session(session, tensor)?);
        }
        let Some(slot) = owned_inputs.get_mut(seed_input_index) else {
            return Err(Error::Internal(format!(
                "semantic eager VJP seed input index {seed_input_index} is outside {input_count} inputs"
            )));
        };
        *slot = Some(copy_value_in_session(session, cotangent_tensor.as_ref())?);
        Ok(())
    })()?;
    let input_refs = owned_inputs
        .iter()
        .enumerate()
        .map(|(index, tensor)| {
            tensor.as_ref().ok_or_else(|| {
                Error::Internal(format!(
                    "semantic eager VJP derivative input {index} was not populated"
                ))
            })
        })
        .collect::<Result<Vec<_>>>()?;
    let input_reads = input_refs
        .iter()
        .map(|tensor| TensorRead::from_tensor(tensor))
        .collect::<Vec<_>>();
    let prepared_runtime = if let Some(prepared_runtime) = prepared_runtime {
        prepared_runtime
    } else {
        let prepared_runtime = Arc::new(ctx.runtime.prepare(
            &execution_program,
            &input_reads,
            std::slice::from_ref(&target),
        )?);
        let entry = Arc::new(PreparedDerivative {
            program: Arc::clone(&derivative_program),
            execution_program: Arc::clone(&execution_program),
            saved_input_indices: saved_input_indices.clone(),
            prepared: Arc::clone(&prepared_runtime),
            seed_input_index,
            derivative_output_indices: derivative_output_indices.clone(),
        });
        ctx.lock_prepared_derivative_cache()?
            .insert(cache_key, entry);
        prepared_runtime
    };
    let mut outputs = run_prepared_on(&ctx.runtime, engine, &prepared_runtime, &input_reads)?
        .into_iter()
        .map(Some)
        .collect::<Vec<_>>();
    let cotangent_trace =
        TracedTensor::from_shared_tensor_value_symbolic_shape(Arc::clone(&cotangent_tensor))?;

    #[cfg(test)]
    EAGER_SEMANTIC_VJP_EXECUTIONS.fetch_add(1, Ordering::Relaxed);

    wrts.iter()
        .zip(wrt_input_indices)
        .map(|(wrt, input_index)| {
            let Some(derivative_output_index) = input_index
                .and_then(|index| derivative_output_indices.get(index).copied().flatten())
            else {
                return Ok(None);
            };
            let result = outputs
                .get_mut(derivative_output_index)
                .and_then(Option::take)
                .ok_or_else(|| {
                    Error::Internal(format!(
                "semantic eager VJP derivative output {derivative_output_index} unavailable"
            ))
                })?;
            let wrt_trace = wrt.semantic_trace.as_ref().ok_or_else(|| {
                Error::Internal("active eager VJP input has no semantic trace".into())
            })?;
            let semantic_trace = derivative_trace_from_frozen_program(
                &source,
                derivative_program.frozen_program(),
                derivative_output_index,
                &[(seed_input_index, Arc::clone(&cotangent_tensor))],
                &[&output_trace, wrt_trace, &cotangent_trace],
                None,
                "semantic_eager_vjp",
            )?;
            Ok(Some(EagerTensor::new_result_with_semantic_trace(
                Arc::clone(ctx),
                eager_val_key(),
                result,
                true,
                output.trace.clone(),
                Some(semantic_trace),
            )?))
        })
        .collect()
}

fn semantic_eager_jvp_optional(
    ctx: &Arc<EagerRuntime>,
    engine: &mut dyn ExecutionEngine,
    output: &EagerTensor,
    wrt: &EagerTensor,
    tangent: &EagerTensor,
) -> Result<Option<Option<EagerTensor>>> {
    if !eager_semantic_vjp_enabled() {
        return Ok(None);
    }
    let (Some(raw_output_trace), Some(wrt_trace)) =
        (output.semantic_trace.as_ref(), wrt.semantic_trace.as_ref())
    else {
        return Ok(None);
    };
    let Some(wrt_key) = wrt_trace.input_key() else {
        return Ok(None);
    };
    if !raw_output_trace.has_attached_input_key(&wrt_key) {
        return Ok(None);
    }

    // First AD request on this output: run the deferred graph analysis once
    // over the whole raw carrier chain before compiling.
    let output_trace = analyze_deferred_semantic_trace(raw_output_trace)?;

    let mut compiler = GraphCompiler::new();
    let source = compile_ad_source(&mut compiler, &output_trace)?;
    if source.output_count() != 1
        || source.input_keys().len() != source.input_count()
        || source.bindings().len() != source.input_count()
    {
        return Ok(None);
    }
    let Some(wrt_input_index) = source.input_key_index(&wrt_key) else {
        return Ok(None);
    };

    let mut active_inputs = vec![false; source.input_count()];
    if let Some(active) = active_inputs.get_mut(wrt_input_index) {
        *active = true;
    } else {
        return Ok(None);
    }
    let ad = AdContext::with_rules_and_transform_cache(
        ctx.semantic_extension_rules.clone(),
        Arc::clone(&ctx.ad_transform_cache),
    );
    let derivative = ad
        .jvp_program(source.frozen_program(), &active_inputs)
        .map_err(|source| {
            Error::runtime_state_source("semantic_eager_jvp", ErrorPhase::GraphBuild, source)
        })?;
    // derivative_input_indices maps source input → derivative seed input.
    let Some(seed_input_index) = derivative
        .derivative_input_indices()
        .get(wrt_input_index)
        .copied()
        .flatten()
    else {
        return Ok(Some(None));
    };
    // derivative_output_indices maps source output → derivative output.
    // There is always exactly one source output (guarded above).
    let Some(derivative_output_index) = derivative
        .derivative_output_indices()
        .first()
        .copied()
        .flatten()
    else {
        return Ok(Some(None));
    };

    let derivative_program = compiler.compile_frozen_program(derivative.frozen())?;
    let tangent_tensor = Arc::new(RetainedValue::from_tensor(tangent.to_tensor(&mut *engine)?));
    let input_count = derivative_program.input_count();
    let mut owned_inputs: Vec<Option<Tensor>> = (0..input_count).map(|_| None).collect();
    // Primal bindings and the seed are staged on the caller's engine.
    (|| -> Result<()> {
        let session: &mut dyn Engine = &mut *engine;
        for (source_input_index, (_, tensor)) in source.bindings().iter().enumerate() {
            let Some(slot) = owned_inputs.get_mut(source_input_index) else {
                return Err(Error::Internal(format!(
                    "semantic eager JVP derivative program has no primal input slot {source_input_index}"
                )));
            };
            *slot = Some(copy_value_in_session(session, tensor)?);
        }
        let Some(slot) = owned_inputs.get_mut(seed_input_index) else {
            return Err(Error::Internal(format!(
                "semantic eager JVP seed input index {seed_input_index} is outside {input_count} inputs"
            )));
        };
        *slot = Some(copy_value_in_session(session, tangent_tensor.as_ref())?);
        Ok(())
    })()?;
    let input_refs = owned_inputs
        .iter()
        .enumerate()
        .map(|(index, tensor)| {
            tensor.as_ref().ok_or_else(|| {
                Error::Internal(format!(
                    "semantic eager JVP derivative input {index} was not populated"
                ))
            })
        })
        .collect::<Result<Vec<_>>>()?;
    let input_reads = input_refs
        .iter()
        .map(|tensor| TensorRead::from_tensor(tensor))
        .collect::<Vec<_>>();
    let outputs = ctx.run_program(engine, &derivative_program, &input_reads)?;
    let output_count = outputs.len();
    let Some(result) = outputs.into_iter().nth(derivative_output_index) else {
        return Err(Error::Internal(format!(
            "semantic eager JVP derivative output index {derivative_output_index} is outside {} outputs",
            output_count
        )));
    };
    let tangent_trace =
        TracedTensor::from_shared_tensor_value_symbolic_shape(Arc::clone(&tangent_tensor))?;
    let semantic_trace = derivative_trace_from_frozen_program(
        &source,
        derivative.frozen(),
        derivative_output_index,
        &[(seed_input_index, Arc::clone(&tangent_tensor))],
        &[&output_trace, wrt_trace, &tangent_trace],
        None,
        "semantic_eager_jvp",
    )?;

    Ok(Some(Some(EagerTensor::new_result_with_semantic_trace(
        Arc::clone(ctx),
        eager_val_key(),
        result,
        true,
        None,
        Some(semantic_trace),
    )?)))
}

fn validate_same_runtime(
    runtime: &Arc<EagerRuntime>,
    tensor: &EagerTensor,
    role: &'static str,
) -> Result<()> {
    if tensor.ctx_id() != runtime.id() {
        return Err(Error::ContextMismatch {
            lhs: runtime.id(),
            rhs: tensor.ctx_id(),
        });
    }
    let _ = role;
    Ok(())
}

fn copy_value_in_session(session: &mut dyn Engine, value: &RetainedValue) -> Result<Tensor> {
    let read = value.tensor_read().map_err(|error| {
        Error::runtime_state_source("copy_value_for_runtime", ErrorPhase::Execution, error)
    })?;
    session.to_contiguous_read(read).map_err(Error::from)
}

fn validate_seed_tensor(op: &'static str, primal: &EagerTensor, seed: &EagerTensor) -> Result<()> {
    if primal.dtype() != seed.dtype() {
        return Err(
            tenferro_tensor::Error::dtype_mismatch(op, primal.dtype(), seed.dtype()).into(),
        );
    }
    if primal.shape() != seed.shape() {
        return Err(
            tenferro_tensor::Error::shape_mismatch(op, primal.shape(), seed.shape()).into(),
        );
    }
    Ok(())
}

/// Eager tensor with reverse-mode autodiff over concrete tensor values.
///
/// This executes each primitive immediately and records a lightweight reverse
/// DAG for `backward()`. Gradients accumulate across repeated `backward()`
/// calls until they are cleared explicitly.
///
/// # Examples
///
/// ```
/// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
///
/// let ctx = EagerRuntime::new()?;
/// let x = EagerTensor::requires_grad_in(Tensor::from_vec_col_major(vec![3], vec![1.0_f64, 2.0, 3.0]).unwrap(), ctx)?;
/// for _ in 0..2 {
///     let loss = x.runtime().with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
///         let squared = s.mul(&x, &x)?;
///         s.reduce_sum(&squared, Some(&[0]))
///     })?;
///     loss.backward(&mut tenferro_cpu::CpuEngine::serial())?;
/// }
///
/// assert_eq!(x.grad()?.unwrap().as_slice::<f64>().unwrap(), &[4.0, 8.0, 12.0]);
/// x.clear_grad()?;
///
/// assert!(x.grad().unwrap().is_none());
/// # Ok::<(), tenferro_ad::Error>(())
/// ```
#[derive(Clone)]
pub struct EagerTensor {
    pub(crate) key: ValueKey<StdTensorOp>,
    pub(crate) trace: Option<EagerTrace>,
    pub(crate) semantic_trace: Option<TracedTensor>,
    pub(crate) requires_grad: bool,
    grad_slot: GradSlot,
    pub(crate) ctx: Arc<EagerRuntime>,
    _record: Arc<EagerTensorRecord>,
}

pub(crate) struct EagerTensorRecord {
    value: Arc<AdValueRecord>,
    key: ValueKey<StdTensorOp>,
    trace: Option<EagerTrace>,
    semantic_trace: Option<TracedTensor>,
    requires_grad: bool,
    grad_slot: GradSlot,
    ctx: Arc<EagerRuntime>,
}

struct EagerTensorParts {
    ctx: Arc<EagerRuntime>,
    key: ValueKey<StdTensorOp>,
    requires_grad: bool,
    trace: Option<EagerTrace>,
    semantic_trace: Option<TracedTensor>,
    value: Arc<AdValueRecord>,
    register_value: bool,
}

impl fmt::Debug for EagerTensor {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("EagerTensor")
            .field("dtype", &self.dtype())
            .field("shape", &self.shape())
            .field("key", &self.key)
            .field("requires_grad", &self.requires_grad)
            .field("has_trace", &self.trace.is_some())
            .field("has_semantic_trace", &self.semantic_trace.is_some())
            .field("ctx_id", &self.ctx_id())
            .finish_non_exhaustive()
    }
}

impl EagerTensor {
    /// Create an untracked eager tensor inside an existing eager context.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0]).unwrap(), ctx)?;
    ///
    /// assert_eq!(x.value()?.as_slice::<f64>().unwrap(), &[1.0, 2.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] when metadata cannot
    /// be registered in the target context, or a typed tensor/backend error
    /// while materializing the source value.
    pub fn from_tensor_in(tensor: Tensor, ctx: Arc<EagerRuntime>) -> Result<Self> {
        Self::new_leaf(ctx, tensor, false)
    }

    /// Create an untracked eager tensor from compact column-major data inside
    /// an existing eager runtime.
    ///
    /// # Errors
    ///
    /// Returns [`Error::TensorRuntime`] with
    /// [`tenferro_tensor::ValidationError::ShapeMismatch`] when the shape and
    /// data length disagree, or with
    /// [`tenferro_tensor::ValidationError::IntegerOverflow`] when shape
    /// arithmetic overflows. Returns [`Error::RuntimeState`] when eager
    /// metadata cannot be registered.
    pub fn from_vec_col_major_in<T: TensorScalar>(
        shape: impl IntoShapeVec,
        data: Vec<T>,
        ctx: Arc<EagerRuntime>,
    ) -> Result<Self> {
        Self::from_tensor_in(Tensor::from_vec_col_major(shape, data)?, ctx)
    }

    /// Create a tracked eager leaf inside an existing eager context.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::requires_grad_in(Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0]).unwrap(), ctx)?;
    ///
    /// assert!(x.grad().unwrap().is_none());
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] when gradient metadata
    /// cannot be registered in the target context, or a typed tensor/backend
    /// error while creating the leaf.
    pub fn requires_grad_in(tensor: Tensor, ctx: Arc<EagerRuntime>) -> Result<Self> {
        Self::new_leaf(ctx, tensor, true)
    }

    pub(crate) fn new_leaf(
        ctx: Arc<EagerRuntime>,
        tensor: Tensor,
        requires_grad: bool,
    ) -> Result<Self> {
        Self::new_leaf_with_session(ctx, tensor, requires_grad, None)
    }

    pub(crate) fn new_leaf_in_session(
        ctx: Arc<EagerRuntime>,
        tensor: Tensor,
        requires_grad: bool,
        session: &mut dyn Engine,
    ) -> Result<Self> {
        Self::new_leaf_with_session(ctx, tensor, requires_grad, Some(session))
    }

    fn new_leaf_with_session(
        ctx: Arc<EagerRuntime>,
        tensor: Tensor,
        requires_grad: bool,
        session: Option<&mut dyn Engine>,
    ) -> Result<Self> {
        let key = eager_val_key();
        // A host-placement tensor needs no backend session: the CPU backend only
        // copies the host buffer for it, so entering a session would add
        // admission, provider exclusion, and session construction without doing
        // any provider work (#1704). Views and backend-family reads keep the
        // session path, and device runtimes keep it for every read.
        let read = TensorRead::from_tensor(&tensor);
        let semantic_tensor =
            match session {
                Some(session) => session.to_contiguous_read(read).map_err(Error::from)?,
                None => match cpu_host_owned_read(&read) {
                    Some(materialized) => materialized.map_err(Error::from)?,
                    None => return Err(Error::unsupported(
                        "EagerTensor::new_leaf",
                        ErrorPhase::Execution,
                        "this tensor needs an engine to become an eager leaf; create it through \
                         EagerSession::constant_from or EagerSession::variable_from",
                    )),
                },
            };
        let semantic_value = Arc::new(RetainedValue::from_tensor(semantic_tensor));
        let semantic_trace = TracedTensor::from_shared_tensor_value_symbolic_shape(semantic_value)?;
        // Deferred materialization: the per-op/leaf global-metadata registry
        // write for the eager tensor key was unreadable after the semantic
        // trace became the sole AD carrier, so it is dropped.
        // ponytail: leaf input-key metadata is still registered by
        // `from_shared_tensor_value_symbolic_shape`; the eager-key entry was
        // vestigial and removed. Add back only if something reads it.
        let value = AdValueRecord::from_tensor(tensor, "EagerTensor::new_leaf")?;
        Self::from_parts(EagerTensorParts {
            ctx,
            key,
            requires_grad,
            trace: None,
            semantic_trace: Some(semantic_trace),
            value,
            register_value: true,
        })
    }

    pub(crate) fn new_result(
        ctx: Arc<EagerRuntime>,
        key: ValueKey<StdTensorOp>,
        tensor: Tensor,
        requires_grad: bool,
        trace: Option<EagerTrace>,
    ) -> Result<Self> {
        Self::new_result_with_semantic_trace(ctx, key, tensor, requires_grad, trace, None)
    }

    pub(crate) fn new_result_with_semantic_trace(
        ctx: Arc<EagerRuntime>,
        key: ValueKey<StdTensorOp>,
        tensor: Tensor,
        requires_grad: bool,
        trace: Option<EagerTrace>,
        semantic_trace: Option<TracedTensor>,
    ) -> Result<Self> {
        let value = AdValueRecord::from_tensor(tensor, "EagerTensor::new_result")?;
        Self::from_parts(EagerTensorParts {
            ctx,
            key,
            requires_grad,
            trace,
            semantic_trace,
            value,
            register_value: true,
        })
    }

    pub(crate) fn new_unregistered_result_with_semantic_trace(
        ctx: Arc<EagerRuntime>,
        key: ValueKey<StdTensorOp>,
        tensor: Tensor,
        requires_grad: bool,
        trace: Option<EagerTrace>,
        semantic_trace: Option<TracedTensor>,
    ) -> Result<Self> {
        let value = AdValueRecord::from_tensor(tensor, "EagerTensor::new_unregistered_result")?;
        Self::from_parts(EagerTensorParts {
            ctx,
            key,
            requires_grad,
            trace,
            semantic_trace,
            value,
            register_value: false,
        })
    }

    pub(crate) fn new_result_value(
        ctx: Arc<EagerRuntime>,
        key: ValueKey<StdTensorOp>,
        value: TensorValue,
        requires_grad: bool,
        trace: Option<EagerTrace>,
        semantic_trace: Option<TracedTensor>,
    ) -> Result<Self> {
        let (group, slot, dtype, shape) = value.try_into_group_parts().map_err(|_| {
            Error::runtime_state(
                "EagerTensor::new_result_value",
                ErrorPhase::Execution,
                "a TensorValue could not be transferred into its allocation group",
            )
        })?;
        let value = AdValueRecord::from_group(group, slot, dtype, shape);
        Self::from_parts(EagerTensorParts {
            ctx,
            key,
            requires_grad,
            trace,
            semantic_trace,
            value,
            register_value: true,
        })
    }

    fn from_parts(parts: EagerTensorParts) -> Result<Self> {
        let EagerTensorParts {
            ctx,
            key,
            requires_grad,
            trace,
            semantic_trace,
            value,
            register_value,
        } = parts;
        let grad_slot = Arc::new(Mutex::new(None));
        if requires_grad {
            ctx.try_register_grad_slot(&key, &grad_slot)?;
        }
        let record = Arc::new(EagerTensorRecord {
            value: Arc::clone(&value),
            key: key.clone(),
            trace: trace.clone(),
            semantic_trace: semantic_trace.clone(),
            requires_grad,
            grad_slot: Arc::clone(&grad_slot),
            ctx: Arc::clone(&ctx),
        });
        if register_value {
            ctx.try_register_value_record(&key, &record)?;
        }

        Ok(Self {
            key,
            trace,
            semantic_trace,
            requires_grad,
            grad_slot,
            ctx,
            _record: record,
        })
    }

    pub(crate) fn new_untracked_result(ctx: Arc<EagerRuntime>, tensor: Tensor) -> Result<Self> {
        let value =
            AdValueRecord::from_untracked_tensor(tensor, "EagerTensor::new_untracked_result")?;
        Ok(Self::new_untracked_value_record(ctx, value, None))
    }

    pub(crate) fn new_untracked_value_result(
        ctx: Arc<EagerRuntime>,
        value: TensorValue,
    ) -> Result<Self> {
        Self::new_untracked_value_result_with_semantic_trace(ctx, value, None)
    }

    pub(crate) fn new_untracked_value_result_with_semantic_trace(
        ctx: Arc<EagerRuntime>,
        value: TensorValue,
        semantic_trace: Option<TracedTensor>,
    ) -> Result<Self> {
        let (group, slot, dtype, shape) = value.try_into_group_parts().map_err(|_| {
            Error::runtime_state(
                "EagerTensor::new_untracked_value_result",
                ErrorPhase::Execution,
                "a TensorValue could not be transferred into its allocation group",
            )
        })?;
        let value = AdValueRecord::from_group(group, slot, dtype, shape);
        Ok(Self::new_untracked_value_record(ctx, value, semantic_trace))
    }

    fn new_untracked_value_record(
        ctx: Arc<EagerRuntime>,
        value: Arc<AdValueRecord>,
        semantic_trace: Option<TracedTensor>,
    ) -> Self {
        let key = eager_val_key();
        let grad_slot = Arc::new(Mutex::new(None));
        let record = Arc::new(EagerTensorRecord {
            value,
            key: key.clone(),
            trace: None,
            semantic_trace: semantic_trace.clone(),
            requires_grad: false,
            grad_slot: Arc::clone(&grad_slot),
            ctx: Arc::clone(&ctx),
        });
        Self {
            key,
            trace: None,
            semantic_trace,
            requires_grad: false,
            grad_slot,
            ctx,
            _record: record,
        }
    }

    pub(crate) fn from_record(record: Arc<EagerTensorRecord>) -> Self {
        Self {
            key: record.key.clone(),
            trace: record.trace.clone(),
            semantic_trace: record.semantic_trace.clone(),
            requires_grad: record.requires_grad,
            grad_slot: Arc::clone(&record.grad_slot),
            ctx: Arc::clone(&record.ctx),
            _record: record,
        }
    }

    /// Detach this tensor from the reverse graph.
    ///
    /// The returned tensor keeps the concrete value but no longer contributes
    /// gradients to the original graph.
    ///
    /// # Errors
    ///
    /// Returns [`Error::TensorRuntime`] when `engine` cannot materialize the
    /// retained value, [`Error::RuntimeStateSource`] when the value has no
    /// readable tensor or symbolic graph metadata registration is unavailable,
    /// and [`Error::RuntimeState`] when the detached record's allocation-group
    /// descriptor is invalid.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::requires_grad_in(Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0]).unwrap(), ctx)?;
    /// let y = x.detach(&mut tenferro_cpu::CpuEngine::serial()).unwrap();
    ///
    /// assert_eq!(y.value()?.as_slice::<f64>().unwrap(), &[1.0, 2.0]);
    /// assert!(y.grad().unwrap().is_none());
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    pub fn detach(&self, engine: &mut dyn Engine) -> Result<Self> {
        let tensor = self.duplicate_value(engine)?;
        let semantic_trace = TracedTensor::from_tensor_symbolic_shape(tensor)?;
        Ok(Self::new_untracked_value_record(
            self.ctx.clone(),
            Arc::clone(&self._record.value),
            Some(semantic_trace),
        ))
    }

    /// Detach this tensor from its graph and re-register it in a different
    /// context as an untracked leaf.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx_a = EagerRuntime::new()?;
    /// let ctx_b = EagerRuntime::new()?;
    /// let x = EagerTensor::requires_grad_in(Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0]).unwrap(), ctx_a)?;
    /// let d = x.detach_into(&mut tenferro_cpu::CpuEngine::serial(), &ctx_b)?;
    ///
    /// assert!(!d.tracks_grad());
    /// assert_eq!(d.ctx_id(), ctx_b.id());
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::RuntimeState`] if the source cannot be materialized or
    /// the target context cannot register its metadata.
    pub fn detach_into(&self, engine: &mut dyn Engine, ctx: &Arc<EagerRuntime>) -> Result<Self> {
        let tensor = self.to_tensor(&mut *engine)?;
        Self::new_leaf_in_session(Arc::clone(ctx), tensor, false, engine)
    }

    /// Borrow the retained value without creating an owner or copy.
    ///
    /// # Errors
    ///
    /// Returns [`Error::RuntimeState`] when the retained allocation-group
    /// descriptor is unavailable or invalid.
    pub fn value(&self) -> Result<ValueGuard<'_>> {
        self._record.value.value("EagerTensor::value")
    }

    /// Explicitly duplicate this value into a fresh standalone allocation.
    ///
    /// # Errors
    ///
    /// Returns [`Error::RuntimeState`] when the retained value or execution
    /// session is unavailable, or a typed host/backend error when the value
    /// cannot be materialized as a contiguous tensor.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let value = EagerTensor::from_tensor_in(
    ///     Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0])?,
    ///     ctx,
    /// )?;
    /// let duplicate = value.duplicate_value(&mut tenferro_cpu::CpuEngine::serial())?;
    /// assert_eq!(duplicate.as_slice::<f64>()?, &[1.0, 2.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    pub fn duplicate_value(&self, engine: &mut dyn Engine) -> Result<Tensor> {
        self.duplicate_value_in_session(engine)
    }

    fn duplicate_host_value(&self) -> Option<Tensor> {
        // A pooled value duplicates through its descriptor view; a caller-owned
        // payload has no such view and duplicates through its own read path, which
        // copies the value while keeping its element type.
        self.value().ok()?.duplicate_host_tensor().ok()
    }

    pub(crate) fn duplicate_value_in_session(&self, session: &mut dyn Engine) -> Result<Tensor> {
        if let Some(tensor) = self.duplicate_host_value() {
            return Ok(tensor);
        }
        let read = self
            ._record
            .value
            .tensor_read("EagerTensor::duplicate_value")?;
        session.to_contiguous_read(read).map_err(Error::from)
    }

    // INVARIANT: the error variants return the unchanged eager handle so a
    // caller can retry ownership extraction without an implicit copy.
    #[allow(clippy::result_large_err)]
    /// Consume this handle and structurally extract its retained allocation.
    ///
    /// A shared handle is returned unchanged as [`IntoValueError::NotUnique`].
    /// Group extraction failures return the unchanged handle and typed group
    /// error; no copy or fallback materialization is attempted.
    ///
    /// # Errors
    ///
    /// Returns [`IntoValueError::NotUnique`] when another handle retains the
    /// value, or [`IntoValueError::Extract`] when structural group extraction
    /// fails because the allocation is aliased or its descriptor is invalid.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let value = EagerTensor::from_tensor_in(
    ///     Tensor::from_vec_col_major(vec![1], vec![3.0_f64])?,
    ///     ctx,
    /// )?;
    /// let owner = value
    ///     .into_value()
    ///     .expect("a uniquely owned value should be extractable");
    /// assert_eq!(owner.as_slice::<f64>()?, &[3.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    pub fn into_value(self) -> std::result::Result<Tensor, IntoValueError<Self>> {
        if Arc::strong_count(&self._record) != 1 {
            return Err(IntoValueError::NotUnique(self));
        }
        let Self { _record, .. } = self;
        let record = match Arc::try_unwrap(_record) {
            Ok(record) => record,
            Err(record) => return Err(IntoValueError::NotUnique(Self::from_record(record))),
        };
        let EagerTensorRecord {
            value,
            key,
            trace,
            semantic_trace,
            requires_grad,
            grad_slot,
            ctx,
        } = record;
        let value = match Arc::try_unwrap(value) {
            Ok(value) => value,
            Err(value) => {
                let record = Arc::new(EagerTensorRecord {
                    value,
                    key,
                    trace,
                    semantic_trace,
                    requires_grad,
                    grad_slot,
                    ctx,
                });
                return Err(IntoValueError::NotUnique(Self::from_record(record)));
            }
        };
        let AdValueRecord {
            container,
            dtype,
            shape,
        } = value;
        let container = match Arc::try_unwrap(container) {
            Ok(container) => container,
            Err(container) => {
                let record = Arc::new(EagerTensorRecord {
                    value: Arc::new(AdValueRecord {
                        container,
                        dtype,
                        shape,
                    }),
                    key,
                    trace,
                    semantic_trace,
                    requires_grad,
                    grad_slot,
                    ctx,
                });
                return Err(IntoValueError::NotUnique(Self::from_record(record)));
            }
        };
        let (group, slot) = match container {
            // A caller-owned payload is handed back to its owner unchanged.
            RetentionContainer::CallerOwned { tensor } => return Ok(*tensor),
            RetentionContainer::Owned { tensor } => return Ok(tensor),
            RetentionContainer::Pooled { group, slot } => (group, slot),
        };
        match group.into_tensor(slot) {
            Ok(tensor) => Ok(tensor),
            Err((group, error)) => {
                let record = Arc::new(EagerTensorRecord {
                    value: Arc::new(AdValueRecord {
                        container: Arc::new(RetentionContainer::Pooled { group, slot }),
                        dtype,
                        shape,
                    }),
                    key,
                    trace,
                    semantic_trace,
                    requires_grad,
                    grad_slot,
                    ctx,
                });
                Err(IntoValueError::Extract {
                    value: Self::from_record(record),
                    error,
                })
            }
        }
    }

    /// Return this tensor's scalar dtype without materializing through
    /// [`value`](Self::value).
    pub fn dtype(&self) -> DType {
        self._record.value.dtype()
    }

    /// Return this tensor's logical shape without materializing through
    /// [`value`](Self::value).
    pub fn shape(&self) -> &[usize] {
        self._record.value.shape()
    }

    /// Borrow this tensor value as a [`TensorRead`].
    ///
    /// This is the preferred borrowed input boundary for executor calls. It
    /// preserves the option to replace eager storage with non-contiguous views
    /// without forcing callers through [`value`](Self::value).
    ///
    /// # Panics
    ///
    /// Panics if a validated eager value record becomes unavailable, which
    /// indicates an internal invariant violation.
    pub fn tensor_read(&self) -> TensorRead<'_> {
        self._record
            .value
            .tensor_read("EagerTensor::tensor_read")
            .expect("validated eager value record")
    }

    /// Materialize this eager tensor as an owned [`Tensor`].
    ///
    /// This is the owned materialization boundary for callers that need a
    /// standalone compact tensor. The operation is fallible because eager
    /// values may be backed by lazy or backend-resident storage.
    ///
    /// # Errors
    ///
    /// Returns [`Error::RuntimeState`] if backend state is unavailable, or a
    /// typed tensor backend error when contiguous materialization fails.
    pub fn to_tensor(&self, engine: &mut dyn Engine) -> Result<Tensor> {
        self.duplicate_value(engine)
    }

    /// Return the accumulated gradient currently stored for this tensor.
    ///
    /// The stored gradient accumulates across repeated `backward()` calls
    /// until it is cleared explicitly.
    ///
    /// For complex scalar losses, stored gradients use tenferro's
    /// Hermitian-adjoint cotangent convention. See
    /// <https://tensor4all.org/tenferro-rs/guides/complex-ad.html>.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::requires_grad_in(Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0]).unwrap(), ctx.clone()).unwrap();
    /// let loss = ctx.with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let y = s.exp(&x)?;
    ///     s.reduce_sum(&y, Some(&[0]))
    /// })?;
    /// let _cotangents = loss.backward(&mut tenferro_cpu::CpuEngine::serial()).unwrap();
    ///
    /// let grad = x.grad()?.unwrap();
    /// assert_eq!(grad.shape(), &[2]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::RuntimeState`] if the gradient slot is poisoned or no
    /// longer available.
    pub fn grad(&self) -> Result<Option<GradientValue>> {
        self.grad_slot
            .lock()
            .map_err(|_| {
                Error::runtime_state(
                    "eager_gradient_slot",
                    ErrorPhase::Execution,
                    "lock poisoned",
                )
            })
            .map(|slot| {
                slot.as_ref().map(|record| GradientValue {
                    record: Arc::clone(record),
                })
            })
    }

    /// Clear the accumulated gradient stored for this tensor.
    ///
    /// This only affects this tensor's gradient slot. Other tensors in the
    /// same context retain their gradients until they are cleared explicitly or
    /// overwritten by later accumulation.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::requires_grad_in(Tensor::from_vec_col_major(vec![3], vec![1.0_f64, 2.0, 3.0]).unwrap(), ctx.clone()).unwrap();
    /// let y = EagerTensor::requires_grad_in(Tensor::from_vec_col_major(vec![3], vec![4.0_f64, 5.0, 6.0]).unwrap(), ctx).unwrap();
    /// let loss = x.runtime().with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///     let product = s.mul(&x, &y)?;
    ///     s.reduce_sum(&product, Some(&[0]))
    /// })?;
    /// let _ = loss.backward(&mut tenferro_cpu::CpuEngine::serial()).unwrap();
    ///
    /// x.clear_grad()?;
    ///
    /// assert!(x.grad()?.is_none());
    /// assert!(y.grad()?.is_some());
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::RuntimeState`] if the gradient slot lock is poisoned.
    pub fn clear_grad(&self) -> Result<()> {
        *self.grad_slot.lock().map_err(|_| {
            Error::runtime_state(
                "eager_gradient_slot",
                ErrorPhase::Execution,
                "lock poisoned",
            )
        })? = None;
        Ok(())
    }

    /// Report whether this tensor participates in gradient tracking.
    ///
    /// Tracked tensors keep a gradient slot in their eager context; untracked
    /// tensors and detached tensors do not.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let plain = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0]).unwrap(), ctx.clone()).unwrap();
    /// let tracked = EagerTensor::requires_grad_in(Tensor::from_vec_col_major(vec![2], vec![3.0_f64, 4.0]).unwrap(), ctx.clone()).unwrap();
    /// let detached = tracked.detach(&mut tenferro_cpu::CpuEngine::serial()).unwrap();
    ///
    /// assert!(!plain.tracks_grad());
    /// assert!(tracked.tracks_grad());
    /// assert!(!detached.tracks_grad());
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    pub fn tracks_grad(&self) -> bool {
        self.requires_grad
    }

    #[cfg(test)]
    fn debug_trace_saved_value_count(&self) -> Option<usize> {
        None
    }

    /// Return the opaque identifier of the context this tensor belongs to.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![1], vec![1.0_f64]).unwrap(), ctx.clone()).unwrap();
    ///
    /// assert_eq!(x.ctx_id(), ctx.id());
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    pub fn ctx_id(&self) -> ContextId {
        self.ctx.id()
    }

    /// Borrow the eager runtime context that owns this tensor.
    pub fn runtime(&self) -> &Arc<EagerRuntime> {
        &self.ctx
    }

    /// Check whether two tensors belong to the same eager context.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![1], vec![1.0_f64]).unwrap(), ctx.clone()).unwrap();
    /// let y = EagerTensor::from_tensor_in(Tensor::from_vec_col_major(vec![1], vec![2.0_f64]).unwrap(), ctx).unwrap();
    ///
    /// assert!(x.same_context(&y));
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    pub fn same_context(&self, other: &Self) -> bool {
        self.ctx_id() == other.ctx_id()
    }

    #[cfg(test)]
    pub(crate) fn standard_graph_op(
        engine: &mut dyn ExecutionEngine,
        inputs: &[&Self],
        build_graph: impl FnOnce(&[TensorInputKey]) -> Result<Arc<Graph<StdTensorOp>>>,
    ) -> Result<Vec<Self>> {
        let Some(first) = inputs.first() else {
            return Err(Error::Internal(
                "standard eager graph op requires at least one input tensor".to_string(),
            ));
        };
        let ctx = Arc::clone(&first.ctx);
        for tensor in inputs.iter().skip(1) {
            if !first.same_context(tensor) {
                return Err(Error::ContextMismatch {
                    lhs: first.ctx_id(),
                    rhs: tensor.ctx_id(),
                });
            }
        }

        let graph_input_keys = (0..inputs.len())
            .map(|_| next_input_key())
            .collect::<Vec<_>>();
        let graph = build_graph(&graph_input_keys)?;
        let initial_data = graph_input_keys
            .iter()
            .zip(inputs.iter())
            .map(|(key, tensor)| {
                Ok((
                    ValueKey::Input(key.clone()),
                    tensor.to_tensor(&mut *engine)?,
                ))
            })
            .collect::<Result<HashMap<_, _>>>()?;
        let execution = ctx.exec_standard_graph_outputs(engine, graph.as_ref(), initial_data)?;
        if execution.outputs.len() != graph.outputs().len() {
            return Err(Error::Internal(format!(
                "standard eager graph op expected {} graph outputs, got {}",
                graph.outputs().len(),
                execution.outputs.len()
            )));
        }

        if !eager_grad_recording_enabled() || !inputs.iter().any(|input| input.requires_grad) {
            return execution
                .outputs
                .into_iter()
                .map(|output| {
                    Self::new_unregistered_result_with_semantic_trace(
                        Arc::clone(&ctx),
                        eager_val_key(),
                        output,
                        false,
                        None,
                        None,
                    )
                })
                .collect();
        }

        let recorded = record_eager_graph_outputs(
            graph.as_ref(),
            &graph_input_keys,
            &execution.outputs,
            inputs,
        )?;
        if recorded.traces.len() != execution.outputs.len() {
            return Err(Error::Internal(format!(
                "standard eager graph op expected {} eager traces, got {}",
                execution.outputs.len(),
                recorded.traces.len()
            )));
        }

        recorded
            .traces
            .into_iter()
            .zip(recorded.semantic_traces)
            .zip(execution.outputs)
            .map(|((trace, semantic_trace), output)| {
                Self::new_result_with_semantic_trace(
                    Arc::clone(&ctx),
                    trace.key,
                    output,
                    trace.requires_grad,
                    trace.trace,
                    semantic_trace,
                )
            })
            .collect()
    }

    /// Run reverse-mode AD from this scalar output.
    ///
    /// Returns the full cotangent map produced by the reverse pass and also
    /// accumulates into `grad()` for tracked eager tensors reachable from this
    /// output.
    ///
    /// For complex scalar outputs, cotangents use tenferro's Hermitian
    /// real-inner-product convention. See
    /// <https://tensor4all.org/tenferro-rs/guides/complex-ad.html>.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::requires_grad_in(Tensor::from_vec_col_major(vec![3], vec![1.0_f64, 2.0, 3.0]).unwrap(), ctx).unwrap();
    /// for _ in 0..2 {
    ///     let loss = x.runtime().with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| {
    ///         let doubled = s.add(&x, &x)?;
    ///         s.reduce_sum(&doubled, Some(&[0]))
    ///     })?;
    ///     loss.backward(&mut tenferro_cpu::CpuEngine::serial())?;
    /// }
    ///
    /// assert_eq!(x.grad().unwrap().unwrap().as_slice::<f64>().unwrap(), &[4.0, 4.0, 4.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::NonScalarGrad`] when this output is not scalar,
    /// [`Error::UnsupportedAdRule`] when a graph operation lacks a reverse rule,
    /// or a typed validation/backend/runtime-state error during the reverse pass.
    pub fn backward(&self, engine: &mut dyn ExecutionEngine) -> Result<Gradients> {
        if !self.shape().is_empty() {
            return Err(Error::NonScalarGrad {
                shape: self.shape().to_vec(),
            });
        }

        let value = self.to_tensor(&mut *engine)?;
        let seed = one_like_tensor(&value, &mut *engine)?;
        self.backward_from_seed(engine, seed)
    }

    /// Run reverse-mode AD from this output with an explicit cotangent seed.
    ///
    /// This is the stateful eager VJP sugar: it returns the cotangent map and
    /// accumulates reachable tracked leaves into their `grad()` slots. Use
    /// [`EagerRuntime::vjp`] when the VJP result should be returned as a
    /// composable eager tensor without touching grad slots.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::new()?;
    /// let x = EagerTensor::requires_grad_in(
    ///     Tensor::from_vec_col_major(vec![2], vec![2.0_f64, 3.0]).unwrap(),
    ///     ctx.clone(),
    /// )?;
    /// let seed = EagerTensor::from_tensor_in(
    ///     Tensor::from_vec_col_major(vec![2], vec![1.0_f64, 2.0]).unwrap(),
    ///     ctx,
    /// )?;
    /// let y = x.runtime().with_eager_session(&mut tenferro_cpu::CpuEngine::serial(), |s| s.mul(&x, &x))?;
    /// y.backward_with(&mut tenferro_cpu::CpuEngine::serial(), &seed)?;
    /// assert_eq!(x.grad()?.unwrap().as_slice::<f64>().unwrap(), &[4.0, 12.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::ContextMismatch`] when `cotangent` belongs to another
    /// eager runtime, [`Error::Validation`] when its shape or dtype is not a
    /// valid seed, [`Error::UnsupportedAdRule`] for an unavailable reverse
    /// rule, or a typed backend/runtime-state error during execution.
    pub fn backward_with(
        &self,
        engine: &mut dyn ExecutionEngine,
        cotangent: &EagerTensor,
    ) -> Result<Gradients> {
        if !self.same_context(cotangent) {
            return Err(Error::ContextMismatch {
                lhs: self.ctx_id(),
                rhs: cotangent.ctx_id(),
            });
        }
        validate_seed_tensor("backward", self, cotangent)?;
        let seed = cotangent.to_tensor(&mut *engine)?;
        self.backward_from_seed(engine, seed)
    }

    fn backward_from_seed(
        &self,
        engine: &mut dyn ExecutionEngine,
        seed: Tensor,
    ) -> Result<Gradients> {
        let cotangent =
            EagerTensor::new_result(Arc::clone(&self.ctx), eager_val_key(), seed, false, None)?;
        let candidate_keys = {
            let mut slots = self.ctx.lock_grad_slots()?;
            let mut keys = Vec::new();
            slots.retain(|key, slot| {
                if slot.upgrade().is_some() {
                    keys.push(key.clone());
                    true
                } else {
                    false
                }
            });
            keys
        };

        let mut targets = Vec::new();
        for key in candidate_keys {
            let Some(record) = self.ctx.value_record(&key)? else {
                continue;
            };
            if !record.requires_grad {
                continue;
            }
            targets.push((key, EagerTensor::from_record(record)));
        }
        let wrts = targets.iter().map(|(_, wrt)| wrt).collect::<Vec<_>>();
        let gradients = semantic_eager_vjp_many(&self.ctx, engine, self, &wrts, &cotangent)?;
        let cotangents = (|| {
            let session: &mut dyn Engine = &mut *engine;
            let mut cotangents = HashMap::new();
            for ((key, _), grad) in targets.into_iter().zip(gradients) {
                let Some(grad) = grad else {
                    continue;
                };
                let tensor = match grad.into_value() {
                    Ok(tensor) => tensor,
                    Err(IntoValueError::NotUnique(handle)) => {
                        handle.duplicate_value_in_session(session)?
                    }
                    // A gradient whose retained layout is a view (for example a
                    // transpose) cannot be extracted as an owned tensor; copy
                    // it to a compact tensor instead.
                    Err(IntoValueError::Extract { value, .. }) => {
                        value.duplicate_value_in_session(session)?
                    }
                };
                cotangents.insert(key, tensor);
            }
            self.ctx.store_grads(&cotangents, session)?;
            Ok::<_, Error>(cotangents)
        })()?;
        Gradients::from_tensors(cotangents)
    }
}

/// Insert a weak registry entry, first dropping dead entries when the insert
/// would otherwise grow the table.
///
/// Dead entries are otherwise removed only when their own key is looked up, so
/// a long-running runtime would keep one entry per value it ever created.
/// Pruning at the growth point keeps the cost amortized O(1) per insert: a
/// sweep runs at most once per capacity doubling.
fn insert_pruning_dead<K: std::hash::Hash + Eq, V>(
    map: &mut HashMap<K, Weak<V>>,
    key: K,
    value: Weak<V>,
) {
    if map.len() == map.capacity() {
        map.retain(|_, entry| entry.strong_count() > 0);
    }
    map.insert(key, value);
}

pub(crate) fn eager_val_key() -> ValueKey<StdTensorOp> {
    ValueKey::Input(next_input_key())
}

pub(crate) struct RecordedEagerTrace {
    pub(crate) key: ValueKey<StdTensorOp>,
    pub(crate) trace: Option<EagerTrace>,
    pub(crate) requires_grad: bool,
}

pub(crate) struct RecordedEagerOutputs {
    pub(crate) traces: Vec<RecordedEagerTrace>,
    pub(crate) semantic_traces: Vec<Option<TracedTensor>>,
}

pub(crate) fn record_eager_outputs_in_session(
    op: &StdTensorOp,
    outputs: &[&Tensor],
    inputs: &[&EagerTensor],
    session: &mut dyn Engine,
) -> Result<RecordedEagerOutputs> {
    let metadata = outputs
        .iter()
        .map(|output| tensor_meta_from_tensor(output))
        .collect();
    record_eager_outputs_inner(op, metadata, inputs, Some(session))
}

pub(crate) fn record_eager_value_outputs_in_session(
    op: &StdTensorOp,
    outputs: &[&TensorValue],
    inputs: &[&EagerTensor],
    session: &mut dyn Engine,
) -> Result<RecordedEagerOutputs> {
    let metadata = outputs
        .iter()
        .map(|output| tensor_meta_from_value(output))
        .collect();
    record_eager_outputs_inner(op, metadata, inputs, Some(session))
}

fn record_eager_outputs_inner(
    op: &StdTensorOp,
    output_metadata: Vec<TensorMeta>,
    inputs: &[&EagerTensor],
    session: Option<&mut dyn Engine>,
) -> Result<RecordedEagerOutputs> {
    let semantic_traces = record_semantic_eager_outputs(op, &output_metadata, inputs, session)?;
    record_eager_outputs_from_metadata(output_metadata, semantic_traces, inputs)
}

fn record_semantic_eager_outputs(
    op: &StdTensorOp,
    output_metadata: &[TensorMeta],
    inputs: &[&EagerTensor],
    mut session: Option<&mut dyn Engine>,
) -> Result<Vec<Option<TracedTensor>>> {
    // Materialize a constant semantic leaf for any untracked input that lost
    // its implicit semantic trace on the active-edge fast path. This keeps
    // "untracked constant feeds tracked AD" working (PyTorch-style: untracked
    // = constant leaf, no gradient flows to it) without re-recording every
    // untracked op at creation time.
    let mut owned_constants = Vec::<TracedTensor>::new();
    for input in inputs {
        if input.semantic_trace.is_none() {
            let value = match session.as_deref_mut() {
                Some(session) => input.duplicate_value_in_session(session)?,
                None => input.duplicate_host_value().ok_or_else(|| {
                    Error::unsupported(
                        "record_semantic_eager_outputs",
                        ErrorPhase::Execution,
                        "a non-host untracked input needs an engine to become a constant leaf",
                    )
                })?,
            };
            owned_constants.push(TracedTensor::from_tensor_symbolic_shape(value)?);
        }
    }
    let mut constants = owned_constants.iter();
    let mut semantic_inputs: Vec<&TracedTensor> = inputs
        .iter()
        .map(|input| {
            input
                .semantic_trace
                .as_ref()
                .unwrap_or_else(|| constants.next().expect("materialized constant"))
        })
        .collect();
    let promotion_plan =
        eager_input_promotion_plan(op, inputs.len(), |index| inputs[index].dtype());
    // Mirror eager execution in the deferred carrier only. The concrete
    // tensors have already been promoted at the execution boundary, so these
    // casts add semantic graph nodes without an eager copy or backend kernel.
    let promoted_semantic_inputs = if semantic_inputs.iter().enumerate().any(|(index, semantic)| {
        semantic.dtype != promotion_plan.target_dtype(index, semantic.dtype)
    }) {
        Some(
            semantic_inputs
                .iter()
                .enumerate()
                .map(|(index, &semantic)| {
                    let target = promotion_plan.target_dtype(index, semantic.dtype);
                    if semantic.dtype == target {
                        Ok(Cow::Borrowed(semantic))
                    } else {
                        semantic.cast(target).map(Cow::Owned)
                    }
                })
                .collect::<Result<Vec<Cow<'_, TracedTensor>>>>()?,
        )
    } else {
        None
    };
    if let Some(promoted_semantic_inputs) = &promoted_semantic_inputs {
        semantic_inputs = promoted_semantic_inputs.iter().map(Cow::as_ref).collect();
    }
    let exact_semantic_inputs = if matches!(op, StdTensorOp::Concatenate { .. }) {
        Some(
            semantic_inputs
                .iter()
                .zip(inputs)
                .map(|(&semantic, input)| {
                    if semantic.is_concrete_shape() {
                        Ok(semantic.clone())
                    } else {
                        semantic.reshape(input.shape())
                    }
                })
                .collect::<Result<Vec<_>>>()?,
        )
    } else {
        None
    };
    if let Some(exact_semantic_inputs) = &exact_semantic_inputs {
        semantic_inputs = exact_semantic_inputs.iter().collect();
    }
    // Deferred materialization (issue #1665 steps 6-7): append only a raw
    // carrier. The runtime helper retains metadata scopes introduced by the
    // promotion/exactification helpers without analyzing this operation.
    let outputs = tenferro_runtime::extension::append_raw_eager_outputs(
        op.clone(),
        &semantic_inputs,
        output_metadata,
    )?;
    Ok(outputs.into_iter().map(Some).collect())
}

#[cfg(test)]
fn record_eager_graph_outputs(
    graph: &Graph<StdTensorOp>,
    graph_input_keys: &[TensorInputKey],
    outputs: &[Tensor],
    inputs: &[&EagerTensor],
) -> Result<RecordedEagerOutputs> {
    let semantic_traces = record_semantic_eager_graph_outputs(graph, graph_input_keys, inputs)?;
    let output_metadata = outputs.iter().map(tensor_meta_from_tensor);
    record_eager_outputs_from_metadata(output_metadata, semantic_traces, inputs)
}

#[cfg(test)]
fn record_semantic_eager_graph_outputs(
    graph: &Graph<StdTensorOp>,
    graph_input_keys: &[TensorInputKey],
    inputs: &[&EagerTensor],
) -> Result<Vec<Option<TracedTensor>>> {
    let Some(semantic_inputs) = inputs
        .iter()
        .map(|input| input.semantic_trace.as_ref())
        .collect::<Option<Vec<_>>>()
    else {
        return Ok(vec![None; graph.outputs().len()]);
    };
    if graph_input_keys.len() != semantic_inputs.len() {
        return Err(Error::Internal(format!(
            "semantic graph recording expected {} input keys, got {}",
            semantic_inputs.len(),
            graph_input_keys.len()
        )));
    }

    let mut values = HashMap::new();
    for (key, tensor) in graph_input_keys.iter().zip(semantic_inputs) {
        values.insert(ValueKey::Input(key.clone()), tensor.clone());
    }

    for op_node in graph.operations() {
        let input_values = op_node
            .inputs
            .iter()
            .map(|input| {
                let key = match input {
                    ValueRef::Local(local_id) => &graph.values()[*local_id].key,
                    ValueRef::External(key) => key,
                };
                values.get(key).cloned().ok_or_else(|| {
                    Error::Internal(format!(
                        "semantic graph recording missing value for {key:?}"
                    ))
                })
            })
            .collect::<Result<Vec<_>>>()?;
        let input_refs = input_values.iter().collect::<Vec<_>>();
        let semantic_outputs = match &op_node.operation {
            StdTensorOp::Extension(ext) => {
                tenferro_runtime::extension::apply(Arc::clone(ext), &input_refs)?
            }
            op => tenferro_runtime::extension::apply_standard_op(op.clone(), &input_refs)?,
        };
        if semantic_outputs.len() != op_node.outputs.len() {
            return Err(Error::Internal(format!(
                "semantic graph recording expected {} outputs for {:?}, got {}",
                op_node.outputs.len(),
                op_node.operation,
                semantic_outputs.len()
            )));
        }
        for (output_id, output) in op_node.outputs.iter().copied().zip(semantic_outputs) {
            values.insert(graph.values()[output_id].key.clone(), output);
        }
    }

    graph
        .outputs()
        .iter()
        .map(|&output_id| {
            let key = &graph.values()[output_id].key;
            values.get(key).cloned().map(Some).ok_or_else(|| {
                Error::Internal(format!(
                    "semantic graph recording missing output for {key:?}"
                ))
            })
        })
        .collect()
}

fn record_eager_outputs_from_metadata(
    output_metadata: impl IntoIterator<Item = TensorMeta>,
    semantic_traces: Vec<Option<TracedTensor>>,
    inputs: &[&EagerTensor],
) -> Result<RecordedEagerOutputs> {
    let output_metadata = output_metadata.into_iter().collect::<Vec<_>>();
    if semantic_traces.len() != output_metadata.len() {
        return Err(Error::Internal(format!(
            "eager recording expected {} semantic traces, got {}",
            output_metadata.len(),
            semantic_traces.len()
        )));
    }
    let requires_grad =
        eager_grad_recording_enabled() && inputs.iter().any(|input| input.requires_grad);
    let trace_count = output_metadata.len();
    let residual_trace = EagerTrace::new(inputs);
    let traces = (0..trace_count)
        .map(|_| RecordedEagerTrace {
            key: eager_val_key(),
            trace: Some(residual_trace.clone()),
            requires_grad,
        })
        .collect();

    Ok(RecordedEagerOutputs {
        traces,
        semantic_traces,
    })
}

fn tensor_meta_from_value(value: &TensorValue) -> TensorMeta {
    TensorMeta::exact(
        value.dtype(),
        value.shape().iter().copied().map(SymDim::from).collect(),
    )
}

#[cfg(test)]
pub(crate) fn zero_like_tensor(input: &Tensor, backend: &mut dyn Engine) -> Result<Tensor> {
    let host = match input.dtype() {
        // A caller-owned payload has no zero-like runtime tensor.
        DType::External(type_id) => {
            return Err(Error::unsupported(
                "zero_like_tensor",
                ErrorPhase::GraphBuild,
                format!(
                    "an externally defined payload ({:?}) has no zero-like runtime tensor",
                    DType::External(type_id)
                ),
            ));
        }
        DType::F32 => Tensor::from_typed::<f32>(TypedTensor::zeros(input.shape().to_vec())?),
        DType::F64 => Tensor::from_typed::<f64>(TypedTensor::zeros(input.shape().to_vec())?),
        DType::I32 => Tensor::from_typed::<i32>(TypedTensor::zeros(input.shape().to_vec())?),
        DType::I64 => Tensor::from_typed::<i64>(TypedTensor::zeros(input.shape().to_vec())?),
        DType::Bool => Tensor::from_typed::<bool>(TypedTensor::from_vec_col_major(
            input.shape().to_vec(),
            vec![false; input.shape().iter().product()],
        )?),
        DType::C32 => Tensor::from_typed::<tenferro_tensor::Complex32>(TypedTensor::zeros(
            input.shape().to_vec(),
        )?),
        DType::C64 => Tensor::from_typed::<tenferro_tensor::Complex64>(TypedTensor::zeros(
            input.shape().to_vec(),
        )?),
    };
    backend
        .upload_host_tensor(TensorRead::from_tensor(&host))
        .map_err(Error::from)
}

pub(crate) fn one_like_tensor(input: &Tensor, session: &mut dyn Engine) -> Result<Tensor> {
    let host = ones_tensor(input.dtype(), input.shape().to_vec())?;
    session
        .upload_host_tensor(TensorRead::from_tensor(&host))
        .map_err(Error::from)
}

#[cfg(test)]
mod tests;