tenferro_ad/

eager.rs

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,
};
use crate::extension_cache::{ExtensionCacheLimits, ExtensionCacheSelector, ExtensionCacheStore};
#[cfg(test)]
use computegraph::graph::Graph;
use computegraph::ValueKey;
#[cfg(test)]
use computegraph::ValueRef;
use tenferro_cpu::{CpuBackend, CpuBackendError, CpuPlacement};
#[cfg(feature = "cuda")]
use tenferro_gpu::cuda::CudaBackend;
#[cfg(feature = "webgpu")]
use tenferro_gpu::webgpu::WebGpuBackend;
#[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, CoreCapabilityBundle, EngineId, ErrorPhase, ExecutionContextIdentity,
    ExtensionModule, GraphCompiler, HardwareClassId, PreparedCompiledGraph, RegistrationIdentity,
    Runtime, RuntimeConfigError, RuntimeConfigSnapshot, RuntimeEpoch, TracedTensor,
};
#[cfg(test)]
use tenferro_tensor::TensorBackend;
#[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 tenferro_tensor::{BackendSession, BackendSessionHost};

#[cfg(feature = "cuda")]
use crate::eager_backend::cuda_runtime_engine_id;
use crate::eager_backend::{
    cpu_runtime_engine_id, cpu_runtime_hardware_class, eager_runtime_for_backend, EagerBackend,
};
#[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};

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

struct CpuRuntimeSelection {
    snapshot: Arc<RuntimeConfigSnapshot>,
    epoch: RuntimeEpoch,
    engine_id: EngineId,
    registration_identity: RegistrationIdentity,
    capabilities: CoreCapabilityBundle,
}

#[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) };
    /// Runtimes whose session callback is running on this thread.
    static EAGER_ENTERED_RUNTIMES: RefCell<Vec<ContextId>> = const { RefCell::new(Vec::new()) };
    #[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)
}

/// The calling thread's `no_grad`/`capture_trace` depths, carried into a
/// backend-session callback.
///
/// A CPU session may run its callback on an executor worker, where the calling
/// thread's thread-local guards are invisible. The callback thread adds these
/// depths for the callback's duration, so a guard held around a session entry
/// governs the operations inside it; guards started inside the callback stay
/// local to it. On the calling thread itself nothing changes.
#[derive(Clone, Copy)]
struct InheritedEagerModes {
    thread: std::thread::ThreadId,
    no_grad: usize,
    capture: usize,
}

impl InheritedEagerModes {
    fn capture() -> Self {
        Self {
            thread: std::thread::current().id(),
            no_grad: EAGER_NO_GRAD_DEPTH.with(Cell::get),
            capture: EAGER_CAPTURE_DEPTH.with(Cell::get),
        }
    }

    /// Apply the captured depths on the current thread until the returned
    /// scope drops, including on unwind.
    fn enter(self) -> InheritedEagerModesScope {
        let inherited = if std::thread::current().id() == self.thread {
            Self {
                no_grad: 0,
                capture: 0,
                ..self
            }
        } else {
            self
        };
        EAGER_NO_GRAD_DEPTH.with(|depth| depth.set(depth.get() + inherited.no_grad));
        EAGER_CAPTURE_DEPTH.with(|depth| depth.set(depth.get() + inherited.capture));
        InheritedEagerModesScope {
            inherited,
            _not_send: PhantomData,
        }
    }
}

/// Marks one runtime's session callback as running on the current thread, so a
/// nested entry into the same runtime from that callback is rejected before it
/// waits on the runtime's own owner lock.
struct EnteredRuntimeScope {
    id: ContextId,
    // Pops this thread's entry, so it must drop where it was created.
    _not_send: PhantomData<Rc<()>>,
}

impl EnteredRuntimeScope {
    fn enter(id: ContextId) -> Self {
        EAGER_ENTERED_RUNTIMES.with(|entered| entered.borrow_mut().push(id));
        Self {
            id,
            _not_send: PhantomData,
        }
    }

    fn any_entered() -> bool {
        EAGER_ENTERED_RUNTIMES.with(|entered| !entered.borrow().is_empty())
    }
}

impl Drop for EnteredRuntimeScope {
    fn drop(&mut self) {
        EAGER_ENTERED_RUNTIMES.with(|entered| {
            let mut entered = entered.borrow_mut();
            if let Some(position) = entered.iter().rposition(|id| *id == self.id) {
                entered.remove(position);
            }
        });
    }
}

struct InheritedEagerModesScope {
    inherited: InheritedEagerModes,
    // Restores this thread's counters, so it must drop where it was created.
    _not_send: PhantomData<Rc<()>>,
}

impl Drop for InheritedEagerModesScope {
    fn drop(&mut self) {
        let InheritedEagerModes {
            no_grad, capture, ..
        } = self.inherited;
        EAGER_NO_GRAD_DEPTH.with(|depth| depth.set(depth.get().saturating_sub(no_grad)));
        EAGER_CAPTURE_DEPTH.with(|depth| depth.set(depth.get().saturating_sub(capture)));
    }
}

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};
/// use tenferro_cpu::CpuBackend;
///
/// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|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};
/// use tenferro_cpu::CpuBackend;
///
/// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|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(&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)
}

fn cpu_runtime_bridge_unsupported(message: impl Into<String>) -> Error {
    Error::unsupported(
        "CpuPlacementBoundEager::refresh_runtime_selection",
        ErrorPhase::Execution,
        message,
    )
}

fn select_cpu_runtime(runtime: &Runtime) -> Result<CpuRuntimeSelection> {
    let snapshot = runtime
        .snapshot()
        .map_err(|source| runtime_state_source("EagerRuntime::runtime_snapshot", source))?;
    let engine_id = cpu_runtime_engine_id()
        .map_err(|source| runtime_config_error("EagerRuntime::cpu_runtime_engine_id", source))?;
    let expected_hardware = cpu_runtime_hardware_class().map_err(|source| {
        runtime_config_error("EagerRuntime::cpu_runtime_hardware_class", source)
    })?;
    let engine = snapshot
        .engine(&engine_id)
        .ok_or_else(|| cpu_runtime_bridge_unsupported("missing CPU runtime engine"))?;
    validate_cpu_runtime_engine(
        engine.context_identity(),
        engine.hardware_class(),
        engine.capabilities(),
        &expected_hardware,
    )?;
    let epoch = snapshot.epoch();
    let registration_identity = engine.registration_identity();
    let capabilities = engine.capabilities().clone();
    Ok(CpuRuntimeSelection {
        snapshot,
        epoch,
        engine_id,
        registration_identity,
        capabilities,
    })
}

fn validate_cpu_runtime_engine(
    context_identity: ExecutionContextIdentity,
    hardware_class: &HardwareClassId,
    capabilities: &CoreCapabilityBundle,
    expected_hardware: &HardwareClassId,
) -> Result<()> {
    if context_identity != ExecutionContextIdentity::of::<CpuBackend>() {
        return Err(cpu_runtime_bridge_unsupported(
            "CPU runtime context mismatch",
        ));
    }
    if hardware_class != expected_hardware {
        return Err(cpu_runtime_bridge_unsupported(
            "CPU runtime hardware mismatch",
        ));
    }
    if capabilities.elementwise().is_none() {
        return Err(cpu_runtime_bridge_unsupported(
            "missing CPU runtime capability: elementwise",
        ));
    }
    if capabilities.reduction().is_none() {
        return Err(cpu_runtime_bridge_unsupported(
            "missing CPU runtime capability: reduction",
        ));
    }
    if capabilities.indexing().is_none() {
        return Err(cpu_runtime_bridge_unsupported(
            "missing CPU runtime capability: indexing",
        ));
    }
    if capabilities.dot_general().is_none() {
        return Err(cpu_runtime_bridge_unsupported(
            "missing CPU runtime capability: dot_general",
        ));
    }
    if capabilities.layout().is_none() {
        return Err(cpu_runtime_bridge_unsupported(
            "missing CPU runtime capability: layout",
        ));
    }
    Ok(())
}

/// 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 {
    /// Generic extension runtime caches.
    pub extensions: CacheStats,
    /// 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};
/// use tenferro_cpu::CpuBackend;
///
/// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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};
/// use tenferro_cpu::CpuBackend;
///
/// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|s| {
///     let squared = s.mul(&x, &x)?;
///     s.reduce_sum(&squared, Some(&[0]))
/// })?;
/// let _gradients = loss.backward()?;
/// 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>,
    ctx: Arc<EagerRuntime>,
}

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) -> 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")?;
                self.ctx
                    .with_execution_session(|session| session.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};
/// use tenferro_cpu::CpuBackend;
///
/// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|s| {
///     let squared = s.mul(&x, &x)?;
///     s.reduce_sum(&squared, Some(&[0]))
/// })?;
/// let gradients = loss.backward()?;
/// 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};
/// use tenferro_cpu::CpuBackend;
///
/// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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 } => {
                let mut reads = group
                    .read_views(std::slice::from_ref(slot))
                    .map_err(|error| {
                        Error::runtime_state_source(op, ErrorPhase::Execution, error)
                    })?;
                reads.pop().ok_or_else(|| {
                    Error::runtime_state(
                        op,
                        ErrorPhase::Execution,
                        "empty allocation-group binding",
                    )
                })
            }
            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
    }
}

/// Placement-selected CPU view of one [`EagerRuntime`].
///
/// The view snapshots the runtime's CPU coordinator/provider bundle and the
/// immutable runtime registration metadata when [`EagerRuntime::on_cpu`] is
/// called. It holds no resource permit while idle and enters one backend
/// session only while [`Self::with_eager_session`] runs. The session exposes
/// core [`BackendSession`] operations on concrete [`Tensor`] values. This
/// bridge deliberately does not expose the eager runtime's linalg, FFT, einsum,
/// or extension-runtime registries.
///
/// The value is intentionally not `Clone`: mutable use makes concurrent
/// session ownership explicit without adding another backend mutex.
///
/// # Examples
///
/// ```rust
/// use tenferro_ad::EagerRuntime;
/// use tenferro_cpu::CpuPlacement;
///
/// let runtime = EagerRuntime::new()?;
/// let cpu = runtime.on_cpu(CpuPlacement::Auto)?;
/// assert_eq!(cpu.runtime_id(), runtime.id());
/// # Ok::<(), tenferro_ad::Error>(())
/// ```
pub struct CpuPlacementBoundEager {
    runtime: Arc<EagerRuntime>,
    backend: CpuBackend,
    snapshot: Arc<RuntimeConfigSnapshot>,
    epoch: RuntimeEpoch,
    engine_id: EngineId,
    registration_identity: RegistrationIdentity,
    capabilities: CoreCapabilityBundle,
}

impl fmt::Debug for CpuPlacementBoundEager {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("CpuPlacementBoundEager")
            .field("runtime_id", &self.runtime.id())
            .field("placement", &self.backend.placement())
            .field("runtime_epoch", &self.epoch)
            .field("engine_id", &self.engine_id)
            .field("registration_identity", &self.registration_identity)
            .finish_non_exhaustive()
    }
}

impl CpuPlacementBoundEager {
    fn refresh_runtime_selection(&mut self) -> Result<()> {
        let current_epoch = self.runtime.runtime.epoch().map_err(|source| {
            runtime_state_source("CpuPlacementBoundEager::refresh_runtime_selection", source)
        })?;
        if current_epoch == self.epoch {
            return Ok(());
        }

        #[cfg(test)]
        CPU_RUNTIME_SELECTION_REFRESHES.fetch_add(1, Ordering::SeqCst);

        let selection = select_cpu_runtime(&self.runtime.runtime)?;
        self.snapshot = selection.snapshot;
        self.epoch = selection.epoch;
        self.engine_id = selection.engine_id;
        self.registration_identity = selection.registration_identity;
        self.capabilities = selection.capabilities;
        Ok(())
    }

    /// Return the identity of the original eager runtime.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::EagerRuntime;
    /// use tenferro_cpu::CpuPlacement;
    ///
    /// let runtime = EagerRuntime::new()?;
    /// let cpu = runtime.on_cpu(CpuPlacement::Auto)?;
    /// assert_eq!(cpu.runtime_id(), runtime.id());
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    pub fn runtime_id(&self) -> ContextId {
        self.runtime.id()
    }

    /// Return the placement requested when this view was created.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::EagerRuntime;
    /// use tenferro_cpu::CpuPlacement;
    ///
    /// let runtime = EagerRuntime::new()?;
    /// let cpu = runtime.on_cpu(CpuPlacement::Auto)?;
    /// assert_eq!(cpu.placement(), CpuPlacement::Auto);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    pub fn placement(&self) -> CpuPlacement {
        self.backend.placement()
    }

    /// Enter one CPU backend session and run core operations through it.
    ///
    /// One call creates one backend session. Tenferro-managed CPU executors
    /// enter once around the closure and core operations reuse that compatible
    /// execution scope. The closure may borrow stack data and need not be
    /// `'static`.
    ///
    /// This phase-2 bridge accepts only core [`BackendSession`] operations. It
    /// does not lock or dispatch the eager runtime's linalg, FFT, einsum, or
    /// extension registries.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Error};
    /// use tenferro_cpu::CpuPlacement;
    /// use tenferro_tensor::{Tensor, TensorRead};
    ///
    /// let runtime = EagerRuntime::new()?;
    /// let mut cpu = runtime.on_cpu(CpuPlacement::Auto)?;
    /// let lhs = Tensor::from_vec_col_major(vec![1], vec![1.0_f64])?;
    /// let rhs = Tensor::from_vec_col_major(vec![1], vec![2.0_f64])?;
    /// let output = cpu.with_eager_session(|session| {
    ///     session
    ///         .add_read(TensorRead::from_tensor(&lhs), TensorRead::from_tensor(&rhs))
    ///         .map_err(Error::from)
    /// })?;
    /// assert_eq!(output.as_slice::<f64>().unwrap(), &[3.0]);
    /// # Ok::<(), Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns the callback's error unchanged. Core backend operations may
    /// report validation, unsupported capability, backend, or runtime-state
    /// failures through that error. Returns `E::from(`[`Error::SessionEntry`]`)`
    /// without running the callback when the backend cannot admit the session,
    /// for example when it is called from inside another session on this
    /// thread ([`tenferro_tensor::SessionEntryError::Reentered`]), and
    /// `E::from` the runtime-selection error when the CPU placement cannot be
    /// refreshed. Use only the borrowed `session` for work inside the scope.
    pub fn with_eager_session<T: Send, E: From<Error> + Send>(
        &mut self,
        f: impl FnOnce(&mut dyn BackendSession) -> std::result::Result<T, E> + Send,
    ) -> std::result::Result<T, E> {
        self.refresh_runtime_selection().map_err(E::from)?;
        match self.backend.with_backend_session(f) {
            Ok(result) => result,
            Err(entry) => Err(E::from(Error::from(entry))),
        }
    }
}

/// 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_cpu::CpuBackend;
/// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
///
/// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|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,
    runtime: Runtime,
    // The backend and its exact runtime engine registration are selected
    // together during construction and remain paired for this runtime's
    // lifetime. The mutex only serializes mutable backend operations.
    backend: Mutex<EagerBackend>,
    // Fixed at construction with the backend/engine pair; extension dispatch
    // can inspect it while holding the borrowed backend session.
    extension_backend_kind: Option<EagerExtensionBackendKind>,
    extension_install_lock: Mutex<()>,
    pub(crate) extension_caches: Mutex<ExtensionCacheStore>,
    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};
/// use tenferro_cpu::CpuBackend;
///
/// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|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;
/// use tenferro_cpu::CpuBackend;
/// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::new()).unwrap();
/// let escaped = ctx.with_eager_session(|session| session).unwrap();
/// let _ = escaped;
/// ```
pub struct EagerSession<'a> {
    runtime: &'a Arc<EagerRuntime>,
    backend: &'a mut dyn BackendSession,
}

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};
    /// use tenferro_cpu::CpuBackend;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|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};
    /// use tenferro_cpu::CpuBackend;
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|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};
    /// use tenferro_cpu::CpuBackend;
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|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};
    /// use tenferro_cpu::CpuBackend;
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|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(|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(|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(|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(|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(|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(|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(|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(|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(|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(|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};
    /// use tenferro_cpu::CpuBackend;
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::new())?;
    /// let converted = ctx.with_eager_session(|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};
    /// use tenferro_cpu::CpuBackend;
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::new())?;
    /// let casted = ctx.with_eager_session(|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};
    /// use tenferro_cpu::CpuBackend;
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::new())?;
    /// let copied = ctx.with_eager_session(|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, self.backend)
    }

    /// Reshape an eager tensor while retaining a separate result owner.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    /// use tenferro_cpu::CpuBackend;
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|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(), self.backend)
    }

    /// Slice an eager tensor with explicit start, limit, and stride per axis.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, SliceConfig, Tensor};
    /// use tenferro_cpu::CpuBackend;
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::new())?;
    /// let y = ctx.with_eager_session(|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, self.backend)
    }

    /// Broadcast an eager tensor into a larger shape on this session.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    /// use tenferro_cpu::CpuBackend;
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|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, self.backend)
    }

    /// 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(|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(|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(|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(|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(|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(),
            },
            self.backend,
        )
    }

    /// 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(|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),
            self.backend,
        )
    }

    /// 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(|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(),
            },
            self.backend,
        )
    }

    /// 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(|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),
            self.backend,
        )
    }

    /// 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(|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(|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};
    /// use tenferro_cpu::CpuBackend;
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|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, self.backend)
    }

    /// 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(|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(|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(|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(|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};
    /// use tenferro_cpu::CpuBackend;
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|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(self.backend)
    }

    /// 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};
    /// use tenferro_cpu::CpuBackend;
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::new())?;
    /// let c = ctx.with_eager_session(|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, self.backend)
    }

    /// 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(|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
            .backend
            .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};
    /// use tenferro_cpu::CpuBackend;
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::new())?;
    /// let x = ctx.with_eager_session(|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, self.backend)
    }

    /// 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};
    /// use tenferro_cpu::CpuBackend;
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|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};
    /// use tenferro_cpu::CpuBackend;
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|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};
    /// use tenferro_cpu::CpuBackend;
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|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(|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(|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(|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(|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(|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(|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(|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(|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(|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};
    /// use tenferro_cpu::CpuBackend;
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::new())?;
    /// let result = ctx.with_eager_session(|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 },
            self.backend,
        )
    }

    /// 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(|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(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(|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(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(|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};
    /// use tenferro_cpu::CpuBackend;
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::new())?;
    /// let result = ctx.with_eager_session(|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,
                self.backend,
            )?;
            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, &rhs], op, self.backend)
    }

    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, self.backend)
    }

    /// 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_cpu::CpuBackend;
    /// use tenferro_ops::std_tensor_op::StdTensorOp;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::new())?;
    /// let y = ctx.with_eager_session(|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, self.backend)
    }

    /// 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 [`EagerRuntime::with_execution_session`].
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, Tensor};
    /// use tenferro_cpu::CpuBackend;
    /// use tenferro_tensor::TensorRead;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::new())?;
    /// let x = Tensor::from_vec_col_major(vec![2], vec![1.0_f64, -2.0])?;
    /// let copy = ctx.with_eager_session(|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 BackendSession {
        &mut *self.backend
    }

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

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

    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(|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 mut caches = self.runtime.lock_extension_caches()?;
        Ok(f(&mut caches))
    }

    pub(crate) fn execute_prepared_extension(
        &mut self,
        executor: &dyn tenferro_runtime::PreparedOperationExecutor,
        inputs: &[TensorRead<'_>],
    ) -> Result<Vec<Tensor>> {
        // The eager owner is already locked; acquire the extension-cache lock
        // second, as in the top-level extension execution region.
        let mut caches = self.runtime.lock_extension_caches()?;
        executor.execute_in_session(self.backend, &mut caches, inputs)
    }
}

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.backend.try_lock() {
            Ok(backend) => {
                debug.field("backend", &*backend);
            }
            Err(_) => {
                debug.field("backend", &"<locked>");
            }
        }
        match self.extension_caches.try_lock() {
            Ok(caches) => {
                debug.field(
                    "extension_cache_stats",
                    &caches.stats(ExtensionCacheSelector::All),
                );
            }
            Err(_) => {
                debug.field("extension_cache_stats", &"<locked>");
            }
        }
        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 {
    pub(crate) fn lock_backend(&self) -> Result<MutexGuard<'_, EagerBackend>> {
        // A thread inside a session must not wait on an owner lock: a callback
        // of this runtime holds it (the wait never returns), or another thread
        // may hold it while waiting for the permit this thread holds (#1946
        // F1). Report the reentry before blocking instead.
        if EnteredRuntimeScope::any_entered()
            || tenferro_tensor::has_active_backend_session()
            || tenferro_cpu::current_cpu_execution() == tenferro_cpu::CpuThreadExecution::Active
        {
            return Err(tenferro_tensor::SessionEntryError::Reentered {
                backend: "EagerRuntime",
            }
            .into());
        }
        let poisoned =
            || Error::runtime_state("eager_backend", ErrorPhase::Execution, "lock poisoned");
        // A shared execution scope already holds the CPU permit, so waiting for
        // the owner could deadlock the same way. Take it only when it is free.
        if tenferro_cpu::current_cpu_execution() == tenferro_cpu::CpuThreadExecution::SharedScope {
            return match self.backend.try_lock() {
                Ok(backend) => Ok(backend),
                Err(std::sync::TryLockError::Poisoned(_)) => Err(poisoned()),
                Err(std::sync::TryLockError::WouldBlock) => {
                    Err(tenferro_tensor::SessionEntryError::Contended {
                        backend: "EagerRuntime",
                        message: "the runtime is in use by another thread while this thread's \
                                  CPU execution scope holds the permit; waiting could deadlock"
                            .to_owned(),
                    }
                    .into())
                }
            };
        }
        // Independent top-level callers wait for the owner and are served in turn.
        self.backend.lock().map_err(|_| poisoned())
    }

    fn lock_extension_caches(&self) -> Result<MutexGuard<'_, ExtensionCacheStore>> {
        self.extension_caches.lock().map_err(|_| {
            Error::runtime_state(
                "eager_extension_caches",
                ErrorPhase::Execution,
                "lock poisoned",
            )
        })
    }

    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 from_backend(backend: EagerBackend) -> Result<Self> {
        Self::from_backend_with_rules_and_cache(
            backend,
            SemanticExtensionRuleSet::default(),
            Arc::new(AdTransformCache::new()),
        )
    }

    fn from_backend_with_rules_and_cache(
        backend: EagerBackend,
        semantic_extension_rules: SemanticExtensionRuleSet,
        ad_transform_cache: Arc<AdTransformCache>,
    ) -> Result<Self> {
        let runtime = eager_runtime_for_backend(&backend)
            .map_err(|source| runtime_config_error("EagerRuntime::from_backend", source))?;
        let extension_backend_kind = match &backend {
            EagerBackend::Cpu(_) => Some(EagerExtensionBackendKind::Cpu),
            #[cfg(test)]
            EagerBackend::Recording(_) => None,
            #[cfg(feature = "cuda")]
            EagerBackend::Cuda(_) => Some(EagerExtensionBackendKind::Cuda),
            #[cfg(feature = "webgpu")]
            EagerBackend::WebGpu(_) => Some(EagerExtensionBackendKind::WebGpu),
        };
        Ok(Self {
            id: ContextId::fresh(),
            runtime,
            backend: Mutex::new(backend),
            extension_backend_kind,
            extension_install_lock: Mutex::new(()),
            extension_caches: Mutex::new(ExtensionCacheStore::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 CPU eager execution context.
    ///
    /// # 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 provider runtime
    /// registration cannot be configured, preserving the underlying
    /// [`RuntimeConfigError`] as the typed error source.
    pub fn new() -> Result<Arc<Self>> {
        Self::with_cpu_backend(CpuBackend::new())
    }

    /// Create a shared eager execution context from a configured CPU backend.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_cpu::CpuBackend;
    /// use tenferro_ad::{EagerRuntime};
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::with_threads(1)?)?;
    /// assert_eq!(std::sync::Arc::strong_count(&ctx), 1);
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::RuntimeStateSource`] when provider runtime
    /// registration cannot be configured, preserving the underlying
    /// [`RuntimeConfigError`] as the typed error source.
    pub fn with_cpu_backend(backend: CpuBackend) -> Result<Arc<Self>> {
        Ok(Arc::new(Self::from_backend(EagerBackend::cpu(backend))?))
    }

    /// Snapshot a placement-selected CPU handle from this eager runtime.
    ///
    /// The eager backend lock is held only long enough to verify the backend
    /// kind and clone its CPU coordinator/provider snapshot. Placement
    /// resolution happens after that guard is dropped. The returned value does
    /// not hold a resource permit or a second runtime/backend mutex while idle.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::EagerRuntime;
    /// use tenferro_cpu::CpuPlacement;
    ///
    /// let runtime = EagerRuntime::new()?;
    /// let cpu = runtime.on_cpu(CpuPlacement::Auto)?;
    /// assert_eq!(cpu.runtime_id(), runtime.id());
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::RuntimeState`] if the eager backend lock is poisoned,
    /// [`Error::Unsupported`] if the runtime is not CPU-backed, or a typed
    /// tensor runtime error retaining [`tenferro_cpu::CpuPlacementError`] when
    /// the requested placement cannot be resolved.
    pub fn on_cpu(self: &Arc<Self>, placement: CpuPlacement) -> Result<CpuPlacementBoundEager> {
        let backend = {
            let backend = self.lock_backend()?;
            backend.cpu_snapshot().ok_or_else(|| {
                Error::unsupported(
                    "EagerRuntime::on_cpu",
                    ErrorPhase::Execution,
                    "the eager runtime is not CPU-backed",
                )
            })?
        };
        let selection = select_cpu_runtime(&self.runtime)?;
        let backend = backend.for_placement(placement).map_err(|source| {
            let error: tenferro_tensor::Error = CpuBackendError::Placement {
                op: "EagerRuntime::on_cpu",
                source,
            }
            .into();
            Error::from(error)
        })?;
        Ok(CpuPlacementBoundEager {
            runtime: Arc::clone(self),
            backend,
            snapshot: selection.snapshot,
            epoch: selection.epoch,
            engine_id: selection.engine_id,
            registration_identity: selection.registration_identity,
            capabilities: selection.capabilities,
        })
    }

    /// Create a shared CPU eager context with explicit AD extension rules.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_cpu::CpuBackend;
    /// use tenferro_ad::{AdContext, EagerRuntime};
    ///
    /// let ad = AdContext::builder().build().unwrap();
    /// let ctx = EagerRuntime::with_cpu_backend_and_ad_context(CpuBackend::new(), &ad)?;
    /// assert_eq!(std::sync::Arc::strong_count(&ctx), 1);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::RuntimeStateSource`] when provider runtime
    /// registration cannot be configured, preserving the underlying
    /// [`RuntimeConfigError`] as the typed error source.
    pub fn with_cpu_backend_and_ad_context(
        backend: CpuBackend,
        ad: &AdContext,
    ) -> Result<Arc<Self>> {
        Ok(Arc::new(Self::from_backend_with_rules_and_cache(
            EagerBackend::cpu(backend),
            ad.semantic_extension_rules().clone(),
            ad.ad_transform_cache(),
        )?))
    }

    /// Create a shared eager execution context from a configured CUDA backend.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_gpu::cuda::CudaBackend;
    /// use tenferro_ad::EagerRuntime;
    ///
    /// let _ctor: fn(CudaBackend) -> tenferro_ad::Result<std::sync::Arc<EagerRuntime>> =
    ///     EagerRuntime::with_cuda_backend;
    /// ```
    #[cfg(feature = "cuda")]
    ///
    /// # Errors
    ///
    /// Returns [`Error::RuntimeStateSource`] when provider runtime
    /// registration cannot be configured, preserving the underlying
    /// [`RuntimeConfigError`] as the typed error source.
    pub fn with_cuda_backend(backend: CudaBackend) -> Result<Arc<Self>> {
        Ok(Arc::new(Self::from_backend(EagerBackend::cuda(backend))?))
    }

    /// Create a shared CUDA eager context with explicit AD extension rules.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{AdContext, EagerRuntime};
    /// use tenferro_gpu::cuda::CudaBackend;
    ///
    /// let _ctor: fn(CudaBackend, &AdContext) -> tenferro_ad::Result<std::sync::Arc<EagerRuntime>> =
    ///     EagerRuntime::with_cuda_backend_and_ad_context;
    /// ```
    #[cfg(feature = "cuda")]
    ///
    /// # Errors
    ///
    /// Returns [`Error::RuntimeStateSource`] when provider runtime
    /// registration cannot be configured, preserving the underlying
    /// [`RuntimeConfigError`] as the typed error source.
    pub fn with_cuda_backend_and_ad_context(
        backend: CudaBackend,
        ad: &AdContext,
    ) -> Result<Arc<Self>> {
        Ok(Arc::new(Self::from_backend_with_rules_and_cache(
            EagerBackend::cuda(backend),
            ad.semantic_extension_rules().clone(),
            ad.ad_transform_cache(),
        )?))
    }

    /// Create a shared eager execution context from a configured WebGPU backend.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::EagerRuntime;
    /// use tenferro_gpu::webgpu::WebGpuBackend;
    ///
    /// let _ctor: fn(WebGpuBackend) -> tenferro_ad::Result<std::sync::Arc<EagerRuntime>> =
    ///     EagerRuntime::with_webgpu_backend;
    /// ```
    #[cfg(feature = "webgpu")]
    ///
    /// # Errors
    ///
    /// Returns [`Error::RuntimeStateSource`] when provider runtime
    /// registration cannot be configured, preserving the underlying
    /// [`RuntimeConfigError`] as the typed error source.
    pub fn with_webgpu_backend(backend: WebGpuBackend) -> Result<Arc<Self>> {
        Ok(Arc::new(Self::from_backend(EagerBackend::webgpu(backend))?))
    }

    /// Create a shared WebGPU eager context with explicit AD extension rules.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{AdContext, EagerRuntime};
    /// use tenferro_gpu::webgpu::WebGpuBackend;
    ///
    /// let _ctor: fn(WebGpuBackend, &AdContext) -> tenferro_ad::Result<std::sync::Arc<EagerRuntime>> =
    ///     EagerRuntime::with_webgpu_backend_and_ad_context;
    /// ```
    #[cfg(feature = "webgpu")]
    ///
    /// # Errors
    ///
    /// Returns [`Error::RuntimeStateSource`] when provider runtime
    /// registration cannot be configured, preserving the underlying
    /// [`RuntimeConfigError`] as the typed error source.
    pub fn with_webgpu_backend_and_ad_context(
        backend: WebGpuBackend,
        ad: &AdContext,
    ) -> Result<Arc<Self>> {
        Ok(Arc::new(Self::from_backend_with_rules_and_cache(
            EagerBackend::webgpu(backend),
            ad.semantic_extension_rules().clone(),
            ad.ad_transform_cache(),
        )?))
    }

    /// Return an opaque identifier for this context.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_cpu::CpuBackend;
    /// use tenferro_ad::{EagerRuntime};
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::new())?;
    /// assert_ne!(ctx.id(), EagerRuntime::with_cpu_backend(CpuBackend::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};
    /// use tenferro_cpu::CpuBackend;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|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
    }

    pub(crate) fn eager_extension_target(&self) -> Result<EagerExtensionTarget> {
        let backend_kind = self.extension_backend_kind.ok_or_else(|| {
            Error::unsupported(
                "EagerRuntime::eager_extension_target",
                ErrorPhase::Execution,
                "the recording backend has no registered eager extension engine",
            )
        })?;
        let engine_id = match backend_kind {
            EagerExtensionBackendKind::Cpu => cpu_runtime_engine_id(),
            #[cfg(feature = "cuda")]
            EagerExtensionBackendKind::Cuda => cuda_runtime_engine_id(),
            #[cfg(feature = "webgpu")]
            EagerExtensionBackendKind::WebGpu => tenferro_gpu::webgpu::webgpu_runtime_engine_id(),
        }
        .map_err(|source| runtime_config_error("EagerRuntime::eager_extension_target", source))?;
        let target = EagerExtensionTarget {
            engine_id,
            backend_kind,
        };
        validate_eager_extension_target(&self.runtime, &target)?;
        Ok(target)
    }

    /// Clear generic extension runtime cache entries.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_cpu::CpuBackend;
    /// use tenferro_ad::{EagerRuntime};
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::new())?;
    /// ctx.clear_extension_caches()?;
    /// assert_eq!(ctx.cache_stats()?.extensions.entries, 0);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] when the extension
    /// cache lock is poisoned.
    pub fn clear_extension_caches(&self) -> Result<()> {
        self.lock_extension_caches()?.clear();
        Ok(())
    }

    /// Clear every cache owned by this eager context.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_cpu::CpuBackend;
    /// use tenferro_ad::{EagerRuntime};
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::new())?;
    /// ctx.clear_caches()?;
    /// assert_eq!(ctx.cache_stats()?.extensions.entries, 0);
    /// 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_extension_caches()?;
        self.clear_ad_transform_caches()?;
        self.clear_prepared_derivative_cache()?;
        Ok(())
    }

    /// Clear prepared derivative program cache entries.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::EagerRuntime;
    /// use tenferro_cpu::CpuBackend;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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_cpu::CpuBackend;
    /// use tenferro_ad::{EagerRuntime};
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::new())?;
    /// let stats = ctx.cache_stats()?;
    /// assert_eq!(stats.extensions.entries, 0);
    /// 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 {
            extensions: self
                .lock_extension_caches()?
                .stats(ExtensionCacheSelector::All),
            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;
    /// use tenferro_cpu::CpuBackend;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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};
    /// use tenferro_cpu::CpuBackend;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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;
    /// use tenferro_cpu::CpuBackend;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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;
    /// use tenferro_cpu::CpuBackend;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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};
    /// use tenferro_cpu::CpuBackend;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(())
    }

    /// Return the extension cache retention limits.
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] if the extension
    /// cache lock is poisoned.
    pub fn extension_cache_limits(&self) -> Result<ExtensionCacheLimits> {
        Ok(self.lock_extension_caches()?.limits())
    }

    /// Replace extension cache retention limits.
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] if the extension
    /// cache lock is poisoned.
    pub fn set_extension_cache_limits(&self, limits: ExtensionCacheLimits) -> Result<()> {
        self.lock_extension_caches()?.set_limits(limits);
        Ok(())
    }

    /// Enter one backend execution session and run provider-neutral operations.
    ///
    /// The callback receives only a lifetime-bound, non-owning backend session.
    /// The backend and its engine registration are fixed when the eager runtime
    /// is constructed. Extension modules are installed separately and remain
    /// available to later extension operations.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::EagerRuntime;
    /// use tenferro_cpu::CpuBackend;
    /// use tenferro_tensor::{Tensor, TensorRead};
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::new())?;
    /// let lhs = Tensor::from_vec_col_major(vec![1], vec![1.0_f64])?;
    /// let rhs = Tensor::from_vec_col_major(vec![1], vec![2.0_f64])?;
    /// let output = ctx.with_execution_session(|session| {
    ///     session.add_read(TensorRead::from_tensor(&lhs), TensorRead::from_tensor(&rhs))
    /// })??;
    /// assert_eq!(output.as_slice::<f64>()?, &[3.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] if the eager backend
    /// lock is poisoned, and [`tenferro_runtime::Error::SessionEntry`] without
    /// running the callback when the backend cannot admit the session.
    /// Backend operations retain their typed tensor/backend errors inside the
    /// callback result.
    pub fn with_execution_session<R: Send>(
        &self,
        f: impl FnOnce(&mut dyn BackendSession) -> R + Send,
    ) -> Result<R> {
        // Lock order: the eager backend owner lock is taken before admission,
        // and admission never waits on this lock while holding a permit.
        let mut backend = self.lock_backend()?;
        let modes = InheritedEagerModes::capture();
        let id = self.id;
        Ok(backend.with_backend_session(move |session| {
            let _modes = modes.enter();
            let _entered = EnteredRuntimeScope::enter(id);
            f(session)
        })?)
    }

    /// Enter a runtime-bound eager session for one or more eager operations.
    ///
    /// Only tensors owned by this runtime may execute on the borrowed session;
    /// the backend lock and the CPU execution permit remain live for the callback.
    /// The CPU backend may run the callback on a worker thread; the calling
    /// thread's [`Self::no_grad`] and [`Self::capture_trace`] guards still govern
    /// it, because the callback inherits their state for its duration. Guards
    /// started inside the callback end with their own scope and never reach the
    /// calling thread.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    /// use tenferro_cpu::CpuBackend;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|session| session.neg(&x))?;
    /// assert_eq!(y.value()?.as_slice::<f64>()?, &[-2.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// The callback's error type only needs `From<tenferro_ad::Error>`, so a
    /// downstream error type flows through with a single `?`:
    ///
    /// ```rust
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    /// use tenferro_cpu::CpuBackend;
    ///
    /// #[derive(Debug)]
    /// enum AppError {
    ///     Tenferro(tenferro_ad::Error),
    ///     Negative,
    /// }
    /// impl From<tenferro_ad::Error> for AppError {
    ///     fn from(error: tenferro_ad::Error) -> Self {
    ///         AppError::Tenferro(error)
    ///     }
    /// }
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::new()).unwrap();
    /// let x = EagerTensor::from_tensor_in(
    ///     Tensor::from_vec_col_major(vec![1], vec![2.0_f64]).unwrap(),
    ///     ctx.clone(),
    /// )
    /// .unwrap();
    /// let squared = ctx.with_eager_session(|session| -> Result<EagerTensor, AppError> {
    ///     let y = session.mul(&x, &x)?;
    ///     let value = y.value()?;
    ///     if value.as_slice::<f64>().map_err(tenferro_ad::Error::from)?[0] < 0.0 {
    ///         return Err(AppError::Negative);
    ///     }
    ///     Ok(y)
    /// });
    /// assert_eq!(squared.unwrap().value().unwrap().as_slice::<f64>().unwrap(), &[4.0]);
    /// ```
    ///
    /// # Errors
    ///
    /// Returns the callback's error unchanged. Before the callback runs,
    /// returns `E::from(`[`Error::RuntimeState`]`)` if the backend lock is
    /// poisoned, or `E::from(`[`tenferro_runtime::Error::SessionEntry`]`)`
    /// when the backend cannot admit the session (for example same-thread
    /// reentry). `T` and `E` must be `Send` while the CPU session may run the
    /// callback on a pool thread.
    pub fn with_eager_session<T: Send, E: From<Error> + Send>(
        self: &Arc<Self>,
        f: impl FnOnce(&mut EagerSession<'_>) -> std::result::Result<T, E> + Send,
    ) -> std::result::Result<T, E> {
        match self.with_execution_session(|backend| {
            f(&mut EagerSession {
                runtime: self,
                backend,
            })
        }) {
            Ok(result) => result,
            Err(entry) => Err(E::from(entry)),
        }
    }

    /// Materialize a host-placement read without entering a backend session.
    ///
    /// Returns `None` when this runtime's backend has no session-free host
    /// materialization path, in which case the caller must enter a session.
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] if the eager backend
    /// lock is poisoned, or the backend's typed materialization error.
    pub(crate) fn to_contiguous_host_read(&self, input: &TensorRead<'_>) -> Result<Option<Tensor>> {
        let backend = self.lock_backend()?;
        match backend.to_contiguous_host_read(input) {
            Some(materialized) => Ok(Some(materialized.map_err(Error::from)?)),
            None => Ok(None),
        }
    }

    // Lock ordering: the eager backend owner is locked first; the
    // extension-cache lock is acquired only after it and remains held through
    // the borrowed session callback.
    /// Run an extension-owned eager operation with a borrowed backend session
    /// and the eager runtime's extension cache store.
    ///
    /// The eager backend owner is locked before the extension-cache lock is
    /// acquired. The callback receives an
    /// [`tenferro_runtime::ExtensionExecutionContext`] so cache access and
    /// backend execution share one lifetime-bound context without exposing the
    /// owning eager backend. The backend and its engine registration remain
    /// fixed for the eager runtime's lifetime.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::EagerRuntime;
    /// use tenferro_cpu::CpuBackend;
    /// use tenferro_tensor::{Tensor, TensorRead};
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::new())?;
    /// let lhs = Tensor::from_vec_col_major(vec![1], vec![1.0_f64])?;
    /// let rhs = Tensor::from_vec_col_major(vec![1], vec![2.0_f64])?;
    /// let output = ctx.with_extension_execution_context(|extension_ctx| {
    ///     extension_ctx
    ///         .backend_mut()
    ///         .add_read(TensorRead::from_tensor(&lhs), TensorRead::from_tensor(&rhs))
    /// })??;
    /// assert_eq!(output.as_slice::<f64>()?, &[3.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] if the eager backend
    /// or extension-cache lock is poisoned. Errors returned by the callback
    /// remain in its result value.
    pub fn with_extension_execution_context<R: Send>(
        &self,
        f: impl FnOnce(
                &mut tenferro_runtime::ExtensionExecutionContext<'_, dyn BackendSession + '_>,
            ) -> R
            + Send,
    ) -> Result<R> {
        let mut backend = self.lock_backend()?;
        let mut extension_cache_guard = self.lock_extension_caches()?;
        let extension_caches: &mut ExtensionCacheStore = &mut extension_cache_guard;
        let modes = InheritedEagerModes::capture();
        let id = self.id;
        Ok(backend.with_backend_session(move |session| {
            let _modes = modes.enter();
            let _entered = EnteredRuntimeScope::enter(id);
            let mut extension_ctx =
                tenferro_runtime::ExtensionExecutionContext::new(session, extension_caches);
            f(&mut extension_ctx)
        })?)
    }

    /// Run a prepared extension executor through the runtime-owned erased
    /// backend context (the native-context path).
    ///
    /// This is the sibling of [`Self::with_extension_execution_context`] for
    /// prepared operations whose executor does not support the scheduler-owned
    /// session but implements the mandatory `execute` bridge. The concrete
    /// backend is exposed as an erased context whose type identity matches the
    /// executor's binding.
    pub(crate) fn with_extension_erased_context<R: Send>(
        &self,
        f: impl FnOnce(&mut tenferro_runtime::ErasedExecutionContext<'_>, &mut ExtensionCacheStore) -> R
            + Send,
    ) -> Result<R> {
        let mut backend = self.lock_backend()?;
        let mut extension_cache_guard = self.lock_extension_caches()?;
        let extension_caches: &mut ExtensionCacheStore = &mut extension_cache_guard;
        let mut erased = backend.erased_context();
        Ok(f(&mut erased, extension_caches))
    }

    /// Block the current thread until backend work submitted by this eager runtime completes.
    ///
    /// CPU runtimes return immediately. CUDA and WebGPU runtimes synchronize
    /// their current backend work queue.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_cpu::CpuBackend;
    /// use tenferro_ad::EagerRuntime;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::new())?;
    /// ctx.synchronize().unwrap();
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`tenferro_runtime::Error::RuntimeState`] if the backend lock is
    /// poisoned, or a typed tensor backend error if synchronization fails.
    pub fn synchronize(&self) -> Result<()> {
        self.lock_backend()?.synchronize().map_err(Error::from)
    }

    /// Owner-context extension fallback used by `extension::apply_eager` when
    /// the extension has no prepared session executor.
    pub(crate) fn exec_extension_outputs_read(
        &self,
        op: &Arc<dyn tenferro_ops::ext_op::ExtensionOp>,
        inputs: &[TensorRead<'_>],
    ) -> Result<Vec<Tensor>> {
        // Lock ordering: the backend lock is held for the input session; the
        // runtime's extension cache locks are acquired only after it.
        let mut backend =
            profile_eager_op_section("exec_extension_outputs_read.lock_backend", || {
                self.lock_backend()
            })?;
        profile_eager_op_section("exec_extension_outputs_read.exec_op", || {
            exec_extension_op_on_tensor_reads(op, inputs, &mut *backend, &self.runtime)
        })
    }

    #[cfg(test)]
    pub(crate) fn exec_standard_graph_outputs(
        &self,
        graph: &Graph<StdTensorOp>,
        initial_data: HashMap<ValueKey<StdTensorOp>, Tensor>,
    ) -> Result<EagerGraphExecution> {
        let mut backend =
            profile_eager_op_section("exec_graph.lock_backend", || self.lock_backend())?;
        let mut all_values = initial_data;

        profile_eager_op_section("exec_graph.with_backend_session", || {
            backend.with_backend_session(|exec| -> 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_cpu::CpuBackend;
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|s| {
    ///     let product = s.mul(&x, &y)?;
    ///     s.reduce_sum(&product, Some(&[0]))
    /// })?;
    /// let _ = loss.backward().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_cpu::CpuBackend;
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|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_cpu::CpuBackend;
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|s| {
    ///     let y = s.exp(&p)?;
    ///     s.reduce_sum(&y, Some(&[0]))
    /// })?;
    /// let _ = loss.backward().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};
    /// use tenferro_cpu::CpuBackend;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|s| s.mul(&x, &x))?;
    /// let dx = ctx.grad(&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>, output: &EagerTensor, wrt: &EagerTensor) -> Result<EagerTensor> {
        self.grad_optional(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};
    /// use tenferro_cpu::CpuBackend;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|s| s.mul(&y, &y))?;
    /// assert!(ctx.grad_optional(&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>,
        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()?;
        let seed = self.with_execution_session(|session| one_like_tensor(&value, session))??;
        let seed = EagerTensor::new_result(Arc::clone(self), eager_val_key(), seed, false, None)?;
        self.vjp_optional(output, wrt, &seed)
    }

    /// Reverse-mode vector-Jacobian product for eager tensors.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    /// use tenferro_cpu::CpuBackend;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|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(&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>,
        output: &EagerTensor,
        wrt: &EagerTensor,
        cotangent: &EagerTensor,
    ) -> Result<EagerTensor> {
        self.vjp_optional(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};
    /// use tenferro_cpu::CpuBackend;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|s| s.mul(&y, &y))?;
    /// assert!(ctx.vjp_optional(&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>,
        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, output, &[wrt], cotangent)?
            .pop()
            .flatten())
    }

    /// Forward-mode Jacobian-vector product for eager tensors.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    /// use tenferro_cpu::CpuBackend;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|s| s.mul(&x, &x))?;
    /// let dy = ctx.jvp(&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>,
        output: &EagerTensor,
        wrt: &EagerTensor,
        tangent: &EagerTensor,
    ) -> Result<EagerTensor> {
        self.jvp_optional(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};
    /// use tenferro_cpu::CpuBackend;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|s| s.mul(&y, &y))?;
    /// assert!(ctx.jvp_optional(&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>,
        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, output, wrt, tangent)? {
            Some(result) => Ok(result),
            None => Ok(None),
        }
    }

    fn store_grads(
        &self,
        cotangents: &HashMap<ValueKey<StdTensorOp>, Tensor>,
        session: &mut dyn BackendSession,
    ) -> 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 {
    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<PreparedCompiledGraph>,
    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: &PreparedCompiledGraph,
    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>,
    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 cache_key = PreparedDerivativeCacheKey {
        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()?));
    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 in one backend session
    // rather than one entry per value.
    ctx.with_execution_session(|session| -> Result<()> {
        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 prepared_runtime = if let Some(prepared_runtime) = prepared_runtime {
        prepared_runtime
    } else {
        let prepared_runtime = Arc::new(
            ctx.runtime
                .prepare_compiled(&execution_program, &input_refs)?,
        );
        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 = ctx
        .runtime
        .run_prepared(&prepared_runtime, &input_refs)?
        .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>,
    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()?));
    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 in one backend session.
    ctx.with_execution_session(|session| -> Result<()> {
        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 outputs = ctx.runtime.run_compiled(&derivative_program, &input_refs)?;
    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 BackendSession,
    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_cpu::CpuBackend;
/// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
///
/// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|s| {
///         let squared = s.mul(&x, &x)?;
///         s.reduce_sum(&squared, Some(&[0]))
///     })?;
///     loss.backward()?;
/// }
///
/// 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_cpu::CpuBackend;
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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_cpu::CpuBackend;
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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 BackendSession,
    ) -> 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 BackendSession>,
    ) -> 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 ctx.to_contiguous_host_read(&read)? {
                Some(materialized) => materialized,
                None => ctx
                    .with_execution_session(|session| session.to_contiguous_read(read))?
                    .map_err(Error::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.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_cpu::CpuBackend;
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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();
    ///
    /// 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) -> Self {
        let semantic_trace = self
            .duplicate_value()
            .ok()
            .and_then(|tensor| TracedTensor::from_tensor_symbolic_shape(tensor).ok());
        Self::new_untracked_value_record(
            self.ctx.clone(),
            Arc::clone(&self._record.value),
            semantic_trace,
        )
    }

    /// Detach this tensor from its graph and re-register it in a different
    /// context as an untracked leaf.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_cpu::CpuBackend;
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx_a = EagerRuntime::with_cpu_backend(CpuBackend::new())?;
    /// let ctx_b = EagerRuntime::with_cpu_backend(CpuBackend::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(&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, ctx: &Arc<EagerRuntime>) -> Result<Self> {
        Self::from_tensor_in(self.to_tensor()?, Arc::clone(ctx))
    }

    /// 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};
    /// use tenferro_cpu::CpuBackend;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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()?;
    /// assert_eq!(duplicate.as_slice::<f64>()?, &[1.0, 2.0]);
    /// # Ok::<(), tenferro_ad::Error>(())
    /// ```
    pub fn duplicate_value(&self) -> Result<Tensor> {
        if let Some(tensor) = self.duplicate_host_value() {
            return Ok(tensor);
        }
        let read = self
            ._record
            .value
            .tensor_read("EagerTensor::duplicate_value")?;
        self.ctx
            .with_execution_session(|session| session.to_contiguous_read(read))?
            .map_err(Error::from)
    }

    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 BackendSession,
    ) -> 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};
    /// use tenferro_cpu::CpuBackend;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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) -> Result<Tensor> {
        self.duplicate_value()
    }

    /// 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_cpu::CpuBackend;
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|s| {
    ///     let y = s.exp(&x)?;
    ///     s.reduce_sum(&y, Some(&[0]))
    /// })?;
    /// let _cotangents = loss.backward().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),
                    ctx: Arc::clone(&self.ctx),
                })
            })
    }

    /// 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_cpu::CpuBackend;
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|s| {
    ///     let product = s.mul(&x, &y)?;
    ///     s.reduce_sum(&product, Some(&[0]))
    /// })?;
    /// let _ = loss.backward().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_cpu::CpuBackend;
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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();
    ///
    /// 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_cpu::CpuBackend;
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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_cpu::CpuBackend;
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(
        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()?)))
            .collect::<Result<HashMap<_, _>>>()?;
        let execution = ctx.exec_standard_graph_outputs(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_cpu::CpuBackend;
    /// use tenferro_ad::{EagerRuntime, EagerTensor, Tensor};
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|s| {
    ///         let doubled = s.add(&x, &x)?;
    ///         s.reduce_sum(&doubled, Some(&[0]))
    ///     })?;
    ///     loss.backward()?;
    /// }
    ///
    /// 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) -> Result<Gradients> {
        if !self.shape().is_empty() {
            return Err(Error::NonScalarGrad {
                shape: self.shape().to_vec(),
            });
        }

        let value = self.to_tensor()?;
        let seed = self
            .ctx
            .with_execution_session(|session| one_like_tensor(&value, session))??;
        self.backward_from_seed(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};
    /// use tenferro_cpu::CpuBackend;
    ///
    /// let ctx = EagerRuntime::with_cpu_backend(CpuBackend::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(|s| s.mul(&x, &x))?;
    /// y.backward_with(&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, 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)?;
        self.backward_from_seed(cotangent.to_tensor()?)
    }

    fn backward_from_seed(&self, 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, self, &wrts, &cotangent)?;
        // Shared-handle duplication and gradient storage share one session.
        let cotangents = self.ctx.with_execution_session(|session| {
            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(
    op: &StdTensorOp,
    outputs: &[&Tensor],
    inputs: &[&EagerTensor],
) -> Result<RecordedEagerOutputs> {
    let output_metadata = outputs
        .iter()
        .map(|output| tensor_meta_from_tensor(output))
        .collect::<Vec<_>>();
    record_eager_outputs_inner(op, output_metadata, inputs, None)
}

pub(crate) fn record_eager_outputs_in_session(
    op: &StdTensorOp,
    outputs: &[&Tensor],
    inputs: &[&EagerTensor],
    session: &mut dyn BackendSession,
) -> 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 BackendSession,
) -> 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 BackendSession>,
) -> 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 BackendSession>,
) -> 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.to_tensor()?,
            };
            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<B: TensorBackend>(
    input: &Tensor,
    backend: &mut B,
) -> 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 BackendSession) -> 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;