fuel-dispatch 0.19.1

// SPDX-License-Identifier: MIT OR Apache-2.0
//! Kernel reference + op parameters + binding table. Phase 7.5 B1+B5.
//!
//! [`KernelRef`] is the uniform function-pointer type that every
//! dispatch wrapper matches. Backend-specific typed kernels live in
//! their backend crates (e.g. `fuel_cpu_backend::byte_kernels`); the
//! *wrapper* functions here in fuel-storage bridge the dispatch-
//! erased `Storage` to those typed kernels by matching on
//! `BackendStorage::Cpu(...)` etc.
//!
//! [`OpParams`] is the typed extras bag — one variant per op family
//! that needs auxiliary data beyond inputs and outputs. Most
//! elementwise ops use `OpParams::None`; reductions carry their
//! reduce dims; conv2d carries kernel/stride/padding; etc.
//!
//! [`KernelBindingTable`] (B5) maps
//! `(OpKind, SmallVec<[DType; N]>, BackendId) -> KernelRef`. The dtype
//! list carries per-operand types — inputs in order, then outputs —
//! so mixed-precision ops (e.g. `Cast: src→dst`) and same-dtype ops
//! (e.g. `Add: [T, T, T]`) share the same key shape. Backends register
//! their dispatch wrappers via the `dispatch::register_*` functions;
//! op-builder methods consult the table at DAG construction time using
//! `Graph::target_backend(id)` as the BackendId key.
//!
//! ## Architecture (cycle-avoidance)
//!
//! - Backend crates (`fuel-cpu-backend`, …): typed kernels on their
//!   concrete storage types (`CpuStorageBytes`, …). No fuel-storage
//!   dependency.
//! - fuel-storage (this crate): dispatch wrappers that match
//!   `BackendStorage::Cpu(...)`, extract the typed storage, and
//!   call the backend's typed kernel. KernelBindingTable lives here.
//! - Backend crates depend on fuel-storage? No — only the wrappers
//!   do. fuel-storage already depends on backend crates for variant
//!   types; this round-trip closes naturally.

use std::collections::HashMap;
use std::sync::Arc;
use std::sync::RwLock;

use fuel_graph::QuantType;
use fuel_graph::registry::FusedOpId;
use fuel_ir::conv::{ParamsConv1D, ParamsConvTranspose1D};
use fuel_ir::dispatch::OpKind;
use fuel_ir::probe::BackendId;
use fuel_ir::{DType, Error, Layout, Result};
use smallvec::SmallVec;

/// Inline capacity for the per-operand dtype list in the binding-table
/// key. 8 covers every op currently in flight without spilling to heap:
/// PagedAttn (q + 2 caches + block_table + context_lens + alibi + out)
/// at 7 entries is the worst case in inference; mixed-precision matmul
/// is ≤ 4. Bumping later is one constant change.
pub type KernelDTypes = SmallVec<[DType; 8]>;

/// One entry of the coverage diagnostic: `(backend, op, operand dtypes)`.
///
/// Named rather than spelled out because the bare 3-tuple appears in a return
/// position where it was unreadable, and `#[allow(type_complexity)]` would
/// have asserted the tuple is fine as-is. It is not — the alias is the fix and
/// the lint was right.
///
/// NOT `BindingKey`: that name is already a distinct `pub enum` in this
/// module (`Static` / …). The compiler caught the collision; the lesson is
/// that a type alias introduced for a lint still needs a name check.
pub type CoverageBinding = (BackendId, OpKind, Vec<DType>);

/// Per-binding capability flags. Today carries one flag — `strided_input`
/// — that signals "this kernel walks input strides explicitly and so
/// can consume non-contiguous input layouts (including stride-0
/// broadcast axes) without auto-Contiguize materializing them first."
/// The executor's contiguize gate consults this to skip the materialize
/// step for capable kernels, so broadcast/transpose/slice layouts can
/// reach the kernel as metadata-only views.
///
/// Default is the conservative all-false; binding sites opt in via
/// [`KernelBindingTable::register_with_caps`]. Forward-extensible by
/// adding fields (no enum/bitflags churn).
#[derive(Copy, Clone, Debug, Default, PartialEq, Eq)]
pub struct KernelCaps {
    /// Kernel handles non-contiguous input layouts directly (consumes
    /// the strides side-channel from `KernelRef::layouts`). When true,
    /// the executor passes non-contiguous inputs through unchanged
    /// instead of running auto-Contiguize first. Inputs with non-zero
    /// `start_offset` still go through auto-Contiguize today (slicing
    /// the device buffer to honor offset is a separate concern).
    pub strided_input: bool,
    /// This kernel BAKES a specific broadcast pattern into its body — a
    /// `broadcast_stride0: required` operand it reads as stride-0 (hoisting
    /// `in[0]` / dropping the bcast-axis stride terms). It is therefore NOT
    /// a generic dense-safe kernel: a dense operand routed here would be
    /// read WRONG (element 0 repeatedly). The realize pick
    /// ([`KernelBindingTable::lookup_with_caps`]) SKIPS such a sibling so a dense operand
    /// is never routed into a broadcast-baked kernel — the exclusion is
    /// order-independent (it does not rest on the generic contract being
    /// registered first). Selecting it CORRECTLY, for an operand genuinely
    /// stride-0 on exactly the required axes, is the layout-aware follow-on
    /// (path 1b) that also teaches the contiguize gate; today the safe
    /// behavior is to exclude it, so a `required`-broadcast contract is a
    /// present-but-never-mis-selected sibling. Defaults `false` — every
    /// existing binding is unaffected (the pick stays first-registered).
    pub requires_broadcast: bool,
}

impl KernelCaps {
    /// All flags off. Equivalent to `Default::default()`; provided as a
    /// const for use in const-context registration tables.
    pub const fn empty() -> Self {
        Self {
            strided_input: false,
            requires_broadcast: false,
        }
    }

    /// Just `strided_input` on. Ergonomic for binary/unary registrations
    /// that opt in to the wrapper-side broadcast path.
    pub const fn strided_input() -> Self {
        Self {
            strided_input: true,
            requires_broadcast: false,
        }
    }
}

use fuel_memory::Storage;

/// Uniform function-pointer signature for per-backend op kernels.
///
/// Inputs are passed as a slice of `&Storage` to handle multi-input
/// ops (binary, ternary, custom). Outputs are passed as a slice of
/// `&mut Storage` to handle multi-output ops (topk, var_mean,
/// custom). Most ops use single-element slices.
///
/// `layouts` is a side-channel that parallels `inputs.append(outputs)`
/// — i.e. `layouts[0..inputs.len()]` are the input layouts in order,
/// followed by `layouts[inputs.len()..]` for the output layouts. This
/// is the load-bearing primitive for stride-aware kernels (broadcast,
/// transpose, gather-from-strided). Kernels that only support
/// contiguous-equal-shape inputs (today's default) can ignore
/// `layouts` because the executor's auto-Contiguize pass guarantees
/// every input is contiguous before this is called; the layout slice
/// then carries `Layout::contiguous(shape)` entries that are useful
/// for shape inference but redundant with `Storage.len_bytes / dtype`.
///
/// `OpParams` carries op-family-specific extras (reduce dims,
/// conv2d geometry, etc.). Most kernels match a specific variant;
/// mismatches are programming bugs that the dispatch resolver
/// must prevent. Pure layout/shape duplication that used to live in
/// `OpParams` (e.g. the old `Reduce::input_layout`) now flows through
/// `layouts` instead.
///
/// **Output Storage is pre-allocated** by the executor before the
/// kernel is called. Kernels write into the pre-allocated bytes;
/// they never allocate.
///
/// ## Multi-output kernels (Option C, Session 5)
///
/// Multi-output ops (e.g. SelectiveScan returning `(y, last_state)`)
/// emit ONE `KernelRef`. The contract:
///
/// - `outputs.len() == 1`. The single `Arc<RwLock<Storage>>` is the
///   producer's *bundled* Storage — its `bundle()` returns
///   `Some(&[OutputView; N])` describing each logical slot's
///   `dtype`/`shape`/`layout`/`byte_offset` inside the bundle's
///   underlying byte buffer.
/// - The kernel writes each logical output into its slot's byte
///   range by acquiring a single write lock on the output and
///   striding by the slot's `byte_offset`. The bundle metadata is
///   the authoritative per-slot spec — `outputs[0].read().unwrap()
///   .bundle().expect("bundled storage").get(slot_idx)` is the
///   canonical access path.
/// - Per-slot dtype tags do NOT travel through the kernel's
///   `KernelDTypes` key (which describes inputs + the bundle's
///   primary dtype only); the bundle metadata IS the per-slot
///   dispatch info.
/// - The bundle is pre-allocated by the executor via
///   `allocate_bundled_storage(device, &output_views_spec)` (see
///   `fuel_backend_contract::storage::allocate_bundled_storage`). Kernels
///   never allocate; they only fill bytes.
///
/// Consumers of multi-output producers are NOT multi-output kernels —
/// they're `Op::View` (zero-copy slot projection) or `Op::ViewOwned`
/// (independent slot buffer), which the executor handles directly
/// without invoking a kernel.
///
/// **Production-correct**: kernels return `Result`, never panic.
pub type KernelRef = fn(
    inputs: &[Arc<RwLock<Storage>>],
    outputs: &mut [Arc<RwLock<Storage>>],
    layouts: &[Layout],
    params: &OpParams,
) -> Result<()>;

/// How many rows a [`OpParams::Matmul`] computes — the data-determined-M
/// state for sparse-MoE / capacity-buffer matmuls. `m` in `OpParams::Matmul`
/// is always the row **capacity** (buffer/stride/size); this says how many
/// of those rows to actually compute (`<= m`), the rest left zeroed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MatmulM {
    /// Compute every one of the `m` rows — the universal static default.
    All,
    /// Compute exactly this many rows of the `m`-row capacity buffer. Set
    /// when the count is known at compile (an input-determined extent) or
    /// after a [`Self::Deferred`] is resolved at execute.
    Rows(usize),
    /// The row count is **data-determined** (a producer binds it mid-pass,
    /// e.g. `Op::NonZeroIndices`'s per-expert token count) — resolve at
    /// execute from `produced_syms`, then patch to [`Self::Rows`].
    Deferred(fuel_ir::DynScalar),
}

/// Typed extras bag passed to kernels via [`KernelRef`]. One variant
/// per op family that needs auxiliary parameters; `None` for
/// parameter-less ops (most elementwise unary and binary).
#[derive(Debug, Clone)]
pub enum OpParams {
    /// Op needs no auxiliary parameters. Used by elementwise
    /// unary (relu, neg, sqr, …), elementwise binary (add, mul,
    /// sub, div, …), shape-only ops (reshape, transpose), etc.
    None,

    /// Reduction (sum, max, mean, …) along specific dims. The input
    /// tensor's [`Layout`] flows through the new `layouts` side-channel
    /// (`layouts[0]`) on `KernelRef`, so this variant carries only the
    /// op-specific extras: which dims to reduce and the keepdim flag.
    ///
    /// `dims` is the sorted list of dims to reduce; `keepdim`
    /// controls whether reduced dims are retained as size-1 in
    /// the output (today fuel-graph never asks for keepdim, but
    /// the field is reserved for the future).
    Reduce { dims: Vec<usize>, keepdim: bool },

    /// Matrix multiplication. Carries the dimensions explicitly
    /// because [`Storage`](fuel_memory::Storage) only holds bytes + dtype;
    /// the kernel needs the batch shape and `(m, n, k)` to walk
    /// inputs and outputs.
    ///
    /// Shape contract: lhs `[..lhs_batch.., m, k]` @
    /// rhs `[..rhs_batch.., k, n]` → out `[..lhs_batch.., m, n]`.
    /// Per-axis the batch dims must either match or follow GQA-style
    /// divisibility (`lhs_dim > rhs_dim && lhs_dim % rhs_dim == 0`)
    /// — the kernel maps each lhs batch slot to the corresponding
    /// rhs slot via `rhs_axis_idx = lhs_axis_idx / n_rep_axis`.
    ///
    /// Equal-batch case: `lhs_batch_dims == rhs_batch_dims`. Rank-2
    /// case: both batch vectors are empty. Both work uniformly.
    ///
    /// Inputs are guaranteed contiguous by the executor's
    /// auto-Contiguize pass; transpose flags don't appear here
    /// because `Op::Transpose` is its own metadata-only op in
    /// fuel-graph.
    Matmul {
        lhs_batch_dims: Vec<usize>,
        rhs_batch_dims: Vec<usize>,
        /// The output/LHS row **capacity** (drives size checks + strides).
        m: usize,
        n: usize,
        k: usize,
        /// How many of the `m` rows to actually compute (data-determined-M
        /// sparse-MoE support). `MatmulM::All` for every ordinary matmul.
        m_compute: MatmulM,
    },

    /// 1D convolution geometry (forward path).
    Conv1D(ParamsConv1D),

    /// 2D convolution geometry (forward path). Carries the tuple-shaped
    /// stride/padding that fuel-graph's `Op::Conv2D` uses (asymmetric
    /// supported), plus the input/weight/output shapes the kernel
    /// needs to walk the multi-index. Storage holds only bytes +
    /// dtype, so spatial shapes flow through OpParams.
    ///
    /// Inputs: `x` shape `[N, Cin, Hin, Win]`, `weight` shape
    /// `[Cout, Cin/groups, Kh, Kw]`, optional `bias` shape `[Cout]`.
    /// Output: `[N, Cout, Hout, Wout]`.
    Conv2D {
        x_shape: [usize; 4],
        w_shape: [usize; 4],
        out_shape: [usize; 4],
        stride: (usize, usize),
        padding: (usize, usize),
        dilation: (usize, usize),
        groups: usize,
    },

    /// 1D transposed-convolution geometry.
    ConvTranspose1D(ParamsConvTranspose1D),

    /// 2D transposed-convolution geometry. Mirrors `OpParams::Conv2D`
    /// in shape (inline fields, asymmetric stride/padding/dilation,
    /// `groups`), with the additional `output_padding` parameter that
    /// transposed conv needs to disambiguate output spatial size.
    ///
    /// Inputs: `x` shape `[N, Cin, Hin, Win]`, `weight` shape
    /// `[Cin, Cout/groups, Kh, Kw]` (note transposed channel order
    /// vs Conv2D), optional `bias` shape `[Cout]`.
    /// Output: `[N, Cout, Hout, Wout]`.
    ConvTranspose2D {
        x_shape: [usize; 4],
        w_shape: [usize; 4],
        out_shape: [usize; 4],
        stride: (usize, usize),
        padding: (usize, usize),
        output_padding: (usize, usize),
        dilation: (usize, usize),
        groups: usize,
    },

    /// Sum-reduce a tensor to a smaller broadcast-compatible target
    /// shape. The kernel left-pads `output_shape` with 1s to match
    /// `input_shape`'s rank, then for each axis: if the padded output
    /// dim equals the input dim that axis carries through; if it's 1
    /// the axis is summed away.
    ReduceSumTo {
        input_shape: Vec<usize>,
        output_shape: Vec<usize>,
    },

    /// Max-reduce a tensor to a smaller broadcast-compatible target
    /// shape — the max-symmetric counterpart of `ReduceSumTo`. Same
    /// axis-alignment rules; per-axis reduction is `max` instead of
    /// `+`.
    ReduceMaxTo {
        input_shape: Vec<usize>,
        output_shape: Vec<usize>,
    },

    /// Backward of `ReduceMaxTo`. Inputs: `(x, upstream)` where
    /// `x.shape == input_shape` and `upstream.shape == output_shape`.
    /// Output: `grad_x` of `input_shape`. The kernel recomputes the
    /// forward max via `input_shape → output_shape` axis alignment,
    /// builds a tie-count mask, and routes upstream back to argmax
    /// positions (fair-share on ties).
    ReduceMaxToBackward {
        input_shape: Vec<usize>,
        output_shape: Vec<usize>,
    },

    /// Multi-head scaled-dot-product attention shape + math params.
    /// `q` is `[B, Hq, Sq, D]`, `k` and `v` are `[B, Hkv, Sk, D]`
    /// (Hkv ≤ Hq, GQA-divisible). Optional 4th input `alibi_slopes`
    /// has shape `[Hq]` (presence is implicit in `inputs.len() == 4`).
    ///
    /// `sk` is the **physical** K/V length axis (the allocated extent —
    /// used for strides + byte-length checks). `k_len` is the
    /// **logical** attended length (`k_len <= sk`): a kernel attends only
    /// the first `k_len` rows of the K/V buffers and bottom-right-aligns
    /// the causal mask at offset `k_len - sq` (Phase D symbolic extents:
    /// flash over a fixed-capacity KV-cache with a runtime live prefix).
    /// For the static path the lowering sets `k_len == sk`, so a kernel
    /// that loops `0..k_len` with the `k_len - sq` causal offset is
    /// byte-identical to the old `0..sk` / `kj > qi` form.
    FlashAttn {
        b: usize,
        hq: usize,
        hkv: usize,
        sq: usize,
        sk: usize,
        d: usize,
        k_len: usize,
        softmax_scale: f32,
        causal: bool,
        window_size_left: Option<usize>,
        window_size_right: Option<usize>,
        softcap: Option<f32>,
    },

    /// Paged-cache attention. `q` is `[B, Hq, Sq, D]`; `k_cache` and
    /// `v_cache` are `[num_blocks, block_size, Hkv, D]`. The 4th input
    /// is `block_table: [B, max_blocks_per_seq]` (U32 — physical block
    /// index per logical position). The 5th is `context_lens: [B]`
    /// (U32 — true context length per sequence). Optional 6th is
    /// `alibi_slopes: [Hq]`.
    PagedAttn {
        b: usize,
        hq: usize,
        hkv: usize,
        sq: usize,
        d: usize,
        block_size: usize,
        max_blocks_per_seq: usize,
        num_blocks: usize,
        softmax_scale: f32,
        softcap: Option<f32>,
    },

    /// Slice along a single dim with explicit start/end/step. The
    /// dim is implicit in the input Layout's relabeling for
    /// multi-dim slice; this variant covers the simple case.
    Slice {
        dim: usize,
        start: usize,
        end: usize,
        step: usize,
    },

    // (Earlier `OpParams::Pad { padding: Vec<(usize, usize)>, fill_bytes: Vec<u8> }`
    // was a speculative multi-dim shape with no consumers. The single-
    // dim shape that Op::Pad actually emits lives at the bottom of this
    // enum; if multi-dim padding lands later, it can extend either
    // shape additively.)
    /// Cast input dtype → target dtype. The target lives on the
    /// output Storage's `dtype` field; this variant signals the
    /// op family without requiring a re-read.
    Cast,

    /// Affine transformation `y = mul * x + add`. Used by
    /// `Tensor::affine` / `scale_and_shift`.
    Affine { mul: f64, add: f64 },

    /// Element-wise clamp: `y = clamp(x, min, max)`.
    Clamp { min: f64, max: f64 },

    /// Element-wise integer power: `y = x.powi(exp)`.
    PowI { exp: i32 },

    /// Concatenate N inputs along one dim. The kernel needs the
    /// outer/inner element counts (product of dims before/after
    /// the concat dim) plus each input's size along the concat
    /// dim — that's all that distinguishes a concat from a
    /// sequence of slab-copies. Order matches `Node::inputs`.
    ///
    /// `axis` is the original concat dim in the output's shape;
    /// kernels that want to walk strided rank-N inputs need it
    /// to build the per-axis stride mask (the outer/inner factoring
    /// alone loses the position info).
    Concat {
        /// Product of output dims before the concat dim.
        outer_count: usize,
        /// Per-input size along the concat dim (length = N inputs).
        input_dim_sizes: Vec<usize>,
        /// Product of output dims after the concat dim.
        inner_count: usize,
        /// Original concat axis index in the output's rank-N shape.
        axis: usize,
    },

    /// Softmax along the last dim. The kernel walks
    /// `outer_count` rows of `last_dim` elements each.
    SoftmaxLastDim { outer_count: usize, last_dim: usize },

    /// Last-dim norm parameters shared by RMS-norm and LayerNorm
    /// (no affine flavor for both today). The OpKind selects which
    /// kernel reads this variant.
    NormLastDim {
        outer_count: usize,
        last_dim: usize,
        eps: f64,
    },

    /// Pick slices along a single dim using a rank-1 U32 index
    /// tensor. The kernel needs the outer/inner counts (dims
    /// before/after the selected axis), the source's selected-dim
    /// size for index bounds checking, and the index count.
    IndexSelect {
        outer_count: usize,
        source_dim_size: usize,
        n_indices: usize,
        inner_count: usize,
    },

    /// N-dimensional gather along `dim`. Source and output shapes
    /// agree on every dim except `dim`; the indices tensor (U32)
    /// has output_shape and supplies the source coord for `dim`
    /// at every output position.
    Gather {
        source_shape: Vec<usize>,
        output_shape: Vec<usize>,
        dim: usize,
    },

    /// Rotary position embedding parameters. The kernel walks
    /// `outer_count` × `seq` × `head_dim` elements; `cos`/`sin`
    /// have `[seq, head_dim]` shape and broadcast across the
    /// outer dims.
    Rope {
        outer_count: usize,
        seq: usize,
        head_dim: usize,
    },

    /// Quantized matmul shape: `A [batch, m, k] @ dequant(W) [n, k] →
    /// out [batch, m, n]`. The weight tensor is a flat U32-typed
    /// byte stream representing `n * k / elements_per_block` blocks
    /// of `quant_type`. Activations are F32 today; output is F32.
    QMatMul {
        quant_type: QuantType,
        batch_count: usize,
        m: usize,
        n: usize,
        k: usize,
    },

    /// Index-add along a single dim with rank-1 U32 indices.
    /// Output is `base` with `src[..., i, ...]` accumulated into
    /// `base[..., indices[i], ...]` for every i ∈ 0..n_indices.
    IndexAdd {
        outer_count: usize,
        base_dim_size: usize,
        n_indices: usize,
        inner_count: usize,
    },

    /// N-dimensional scatter-add. Indices and src share the same
    /// shape; base may differ only along `dim`. The kernel walks
    /// every src/indices position, reads `indices[p]` for the
    /// destination's `dim` coord, and accumulates `src[p]` into
    /// `base` at that destination.
    ScatterAdd {
        base_shape: Vec<usize>,
        src_shape: Vec<usize>,
        dim: usize,
    },

    /// Flip the order of elements along one dim. The flat-3-axis
    /// view (outer × dim × inner) lets the kernel walk a tight loop
    /// without re-deriving the axis split per call. `axis` is the
    /// original dim index in the input's rank-N shape — needed by
    /// stride-aware kernels to build a per-axis flip mask.
    Flip {
        outer_count: usize,
        dim_size: usize,
        inner_count: usize,
        axis: usize,
    },

    /// Cyclic shift along one dim by `shift` positions (positive
    /// shifts move elements to higher indices, wrapping around).
    /// Same flat-3-axis view as `Flip`. `shift` is signed: negative
    /// shifts move elements to lower indices.
    Roll {
        outer_count: usize,
        dim_size: usize,
        inner_count: usize,
        shift: i64,
        axis: usize,
    },

    /// Running cumulative sum along one dim. Same flat-3-axis view
    /// as `Flip`/`Roll`. Output is always the same dtype as input;
    /// kernel needs typed addition so it's per-dtype (unlike Flip/Roll).
    CumSum {
        outer_count: usize,
        dim_size: usize,
        inner_count: usize,
        axis: usize,
    },

    /// Triangular mask parameters (used by both Triu and Tril — the
    /// op-kind picks the direction). `batch_count` is the product of
    /// leading dims; `rows`/`cols` are the last two dims. `diagonal`
    /// is signed (0 = main diagonal, positive shifts up, negative
    /// shifts down).
    Triangular {
        batch_count: usize,
        rows: usize,
        cols: usize,
        diagonal: i64,
    },

    /// LogSoftmax along the last dim. Walks `outer_count` rows of
    /// `last_dim` elements each. Per-dtype kernel (uses log/exp).
    LogSoftmaxLastDim { outer_count: usize, last_dim: usize },

    /// MaskedFill: per-element fill where mask is nonzero. The kernel
    /// reads the element count from the layout; `fill_bytes` is
    /// pre-encoded in the output's dtype (one element's worth).
    MaskedFill { fill_bytes: Vec<u8> },

    /// Backward helper for Pad. Carries the input shape so the kernel
    /// can size its scatter-add buffer; the output shape is implicit
    /// from the input shape + padding.
    PadBackward {
        in_shape: Vec<usize>,
        out_shape: Vec<usize>,
        padding: Vec<(usize, usize)>,
        mode_tag: u8,
    },

    /// Multi-dim Pad: per-axis (before, after) plus a `mode_tag`
    /// (0=Constant, 1=Reflect, 2=Replicate) and pre-encoded
    /// `fill_bytes` for Constant fill. Dtype-agnostic at the byte
    /// level: the kernel just copies bytes per the input/output shapes,
    /// which is why fill is bytes (already encoded in the output's
    /// dtype) rather than `f64` (which would force the kernel to
    /// know its dtype).
    ///
    /// `in_shape.len() == out_shape.len() == padding.len()`, and
    /// `out_shape[i] == in_shape[i] + padding[i].0 + padding[i].1`.
    /// `fill_bytes.len() == dtype_size_in_bytes` (one element's worth).
    Pad {
        in_shape: Vec<usize>,
        out_shape: Vec<usize>,
        padding: Vec<(usize, usize)>,
        mode_tag: u8,
        fill_bytes: Vec<u8>,
    },

    /// In-place scatter write parameters. The destination shape +
    /// source shape together determine the kernel walk; `ranges`
    /// gives the per-axis (start, end) slab inside the destination.
    /// `dest_shape[i] >= ranges[i].1`, and source shape is implicitly
    /// `ranges[i].1 - ranges[i].0` along axis `i`.
    ///
    /// Phase E.3.2: backs `Op::WriteSlice` for persistent KV-cache
    /// writes.
    ///
    /// `deferred_dyn_offset` is the **data-determined dynamic-shapes** seam
    /// (increment 2b). Normally `None` — `ranges` is already resolved at
    /// compile (static, or an *input-determined* `dyn_offset` bound in the
    /// per-pass `SymEnv`). It is `Some((axis, sym))` only when the offset is
    /// *data-determined* — its `SymId` is UNBOUND at compile because a
    /// producer (e.g. `Op::NonZeroIndices`) computes it mid-pass. The
    /// `WorkItemKind::WriteSlice` executor arm resolves it from
    /// `produced_syms` and overrides `ranges[axis]` to `(base, base+width)`
    /// just before the kernel runs; the kernel wrappers ignore this field.
    WriteSlice {
        dest_shape: Vec<usize>,
        ranges: Vec<(usize, usize)>,
        deferred_dyn_offset: Option<(usize, fuel_ir::DynScalar)>,
    },

    /// In-place ring-buffer scatter write parameters. Like
    /// `WriteSlice` except `axis` wraps modulo `modulus`; the
    /// dynamic start on that axis is read from an extra input (rank-0
    /// U32). On the rotating axis, `ranges[axis].0` is ignored
    /// (dynamic) and `ranges[axis].1 - ranges[axis].0` is the slab
    /// width (must equal the source dim on that axis, and must not
    /// exceed `modulus`).
    ///
    /// Phase C: backs `Op::WriteSliceRotating` for sliding-window
    /// KV caches.
    WriteSliceRotating {
        dest_shape: Vec<usize>,
        axis: usize,
        modulus: usize,
        ranges: Vec<(usize, usize)>,
    },

    /// In-place scatter write whose start on `axis` is read from a
    /// **device-resident** rank-0 `I64` input (`inputs[2]`) at kernel
    /// launch — the CUDA-graph-capturable KV-cache append. Like
    /// `WriteSlice` except `ranges[axis].0` is a placeholder overridden
    /// device-side by the offset; no modulo wrap. The CUDA wrapper
    /// passes the offset buffer's device pointer straight through as
    /// `dyn_start_dev` (NO D2H — that would break capture); the CPU
    /// wrapper reads it host-side and delegates to `write_slice_cpu`.
    /// Bounds (`offset + width <= dest_shape[axis]`) are the caller's
    /// contract; the kernel does not clamp.
    ///
    /// CapturedRun (2026-07): backs `Op::WriteSliceDoff` for
    /// `DecodeSession` decode-step capture/replay.
    WriteSliceDoff {
        dest_shape: Vec<usize>,
        axis: usize,
        ranges: Vec<(usize, usize)>,
    },

    /// FusedSoftmaxCrossEntropy execution parameters. Inputs:
    /// `logits [n_rows, vocab]` (F32, flattened from the original
    /// `[..., V]` shape) and `targets [n_rows]` (I64). The kernel
    /// walks row-by-row, computing stable log-softmax + NLL +
    /// `ignore_index` masking + the requested reduction in a single
    /// pass, allocating only an `[n_rows]` per-row accumulator (and
    /// a scalar for Mean/Sum). Output is F32 scalar `[]` for
    /// Mean/Sum, F32 `[n_rows]` for None.
    ///
    /// `n_rows` and `vocab` are derived at translate time from the
    /// logits layout; the kernel uses them to iterate without
    /// re-parsing shapes.
    FusedSoftmaxCrossEntropy {
        n_rows: usize,
        vocab: usize,
        reduction: fuel_graph::registry::Reduction,
        ignore_index: i64,
    },

    /// CausalConv1d execution parameters. Inputs:
    /// `x [batch, channels, seq_in]` (pre-padded by caller with
    /// `kernel - 1` left zeros), `weight [channels, 1, kernel]`,
    /// `bias [channels]`. Output `[batch, channels, seq_out]` where
    /// `seq_out = seq_in - (kernel - 1)`. All tensors share dtype.
    ///
    /// Carries both `seq_in` and `seq_out` so the kernel walks the
    /// time axis without re-deriving from shapes. `use_silu` toggles
    /// the fused SiLU activation on the output store (matches
    /// baracuda's `causal_conv1d_*_run` signature flag).
    CausalConv1d {
        batch: usize,
        channels: usize,
        seq_in: usize,
        seq_out: usize,
        kernel: usize,
        use_silu: bool,
    },

    /// SelectiveScan execution parameters. 5 required inputs:
    /// `u [batch, seqlen, dim]`, `delta [batch, seqlen, dim]`,
    /// `a [dim, dstate]`, `b [batch, seqlen, dstate]`,
    /// `c [batch, seqlen, dstate]`. Output `y [batch, seqlen, dim]`.
    /// All tensors share dtype.
    ///
    /// `delta_softplus` toggles applying softplus(delta) before use
    /// (matches baracuda's `selective_scan_*_run` flag).
    SelectiveScan {
        batch: usize,
        seqlen: usize,
        dim: usize,
        dstate: usize,
        delta_softplus: bool,
    },

    /// SsdChunkScan execution parameters. 5 inputs:
    /// `x [batch, seqlen, heads, head_dim]`,
    /// `dt [batch, seqlen, heads]`, `a [heads]`,
    /// `b [batch, seqlen, heads, state_dim]`,
    /// `c [batch, seqlen, heads, state_dim]`. Output `y` matches `x`'s
    /// shape. All tensors share dtype.
    ///
    /// `chunk_size` is the SSD block size (GPU-parallelism knob).
    /// On CPU the kernel runs a sequential scan regardless;
    /// validation requires `chunk_size > 0` and
    /// `seqlen % chunk_size == 0`.
    SsdChunkScan {
        batch: usize,
        seqlen: usize,
        heads: usize,
        head_dim: usize,
        state_dim: usize,
        chunk_size: usize,
    },

    /// Nf4Matmul execution parameters. 3 inputs:
    /// `activations [batch, m, k]` (rank ≥ 2; leading dims flattened
    /// into `batch`), `w_packed [n, k/2]` U8, `absmax [n, k/block_size]`
    /// F32. Output `[batch, m, n]` matching activations' dtype.
    ///
    /// `block_size` is the per-output-row, per-block scale granularity
    /// (typically 64 in bitsandbytes). `k` must be even (w_packed
    /// layout requirement) and a multiple of `block_size`.
    Nf4Matmul {
        batch: usize,
        m: usize,
        n: usize,
        k: usize,
        block_size: usize,
    },

    /// A **runtime fused op's** live scalar args (the `extract:` values from
    /// `FusedOpParams::Runtime { scalars }`), in the region's pattern
    /// pre-order — the same canonical slot order the synthesized kernel's
    /// trailing `, float p0, float p1, …` launch parameters are declared in
    /// (the synthesizer's scalar ABI appends them after `long long n`, always
    /// `float`-typed regardless of the kernel's element dtype). Only produced
    /// by `compile_one`'s `is_runtime` arm; a param-free runtime op carries
    /// `OpParams::None` instead.
    JitScalars {
        /// Slot values, pattern pre-order. `f64` at rest (the graph-side
        /// carrier); the CUDA launch narrows each to `f32` per the ABI.
        scalars: Vec<f64>,
    },

    /// NonZeroIndices execution parameters. 1 input `x` (the value
    /// tensor); one bundled output `[indices [capacity] U32 ; count
    /// [1] U32]` where `capacity == x.elem_count()`. The kernel writes
    /// the ascending flat indices of `x`'s nonzero elements into slot 0
    /// (first `count` entries) and the runtime `count` into slot 1.
    ///
    /// `count_sym` is the [`fuel_ir::symbol::SymId`] the executor binds to the realized
    /// `count` in the per-pass `SymEnv` *after* this op runs — the
    /// data-determined dynamic-shape seam. The kernel itself ignores it.
    NonZeroIndices {
        capacity: usize,
        count_sym: fuel_ir::SymId,
    },
}

// =============================================================================
// Phase 7.5 B5 — kernel binding table
// =============================================================================

/// Maps `(OpKind, KernelDTypes, BackendId)` triples to dispatch wrapper
/// functions, where `KernelDTypes` lists per-operand dtypes (inputs in
/// order, then outputs). Built once at backend registration time
/// (typically a process-wide `OnceLock` though that's not enforced
/// here for testability), consulted at execute time via lookup.
///
/// Same-dtype ops register `[T, T, ..., T]` for the right operand
/// count; mixed-precision ops (Cast, future F32×BF16→F32 matmul)
/// register the exact combo. Variadic ops (Concat) register a
/// canonical short shape `[T, T]` (one input dtype + output dtype);
/// the lookup site for those ops collapses its dtypes vector to the
/// same shorthand.
///
/// Backends register their wrappers via the `dispatch::register_*`
/// functions in this crate (e.g.
/// [`crate::dispatch::register_cpu_kernels`]).
/// Cost-fn signature for primitive-op registrations stored in
/// [`KernelBindingTable`]. Mirrors the fused-op cost-fn shape but
/// takes [`OpParams`] (the binding-table param payload) instead of
/// `FusedOpParams`.
///
/// Implementations return a [`crate::fused::CostEstimate`] computed
/// statically from shapes + dtypes + op-specific params + backend
/// capabilities. The architecture's Layer-1 (FLOP-count + bandwidth)
/// cost model lives here; Layer-2 empirical refinement composes on
/// top via the telemetry framework, not by changing this signature.
pub type CostFn = fn(
    &[fuel_ir::Shape],
    &[DType],
    &OpParams,
    &fuel_ir::backend::BackendCapabilities,
) -> crate::fused::CostEstimate;

/// Sentinel "no cost claim" function — returns
/// [`crate::fused::CostEstimate::default`] (all-zero). The fill pass
/// (`fill_unset_cpu_cost`) recognizes this exact function pointer
/// to decide which entries get the OpKind-family default.
pub fn unknown_cost(
    _shapes: &[fuel_ir::Shape],
    _dtypes: &[DType],
    _params: &OpParams,
    _caps: &fuel_ir::backend::BackendCapabilities,
) -> crate::fused::CostEstimate {
    crate::fused::CostEstimate::default()
}

/// One concrete kernel registered against a `(OpKind, dtypes, backend)`
/// decision-point key. Phase 7.6 step 9a: a single key now carries
/// `SmallVec<[BindingEntry; 2]>` — multiple alternatives compete at
/// the same decision point (e.g. cuBLAS bf16 matmul + CUTLASS bf16
/// matmul at `(MatMul, [BF16, BF16, BF16], Cuda)`). The route picker
/// (step 9b) selects among them at plan time; today's single-impl
/// callers (`lookup`, `lookup_with_caps`, `lookup_precision`,
/// `lookup_cost`) return the first entry, preserving the pre-9a
/// behavior for sites that haven't migrated.
#[derive(Clone, Copy, Debug)]
pub struct BindingEntry {
    pub kernel: KernelRef,
    pub caps: KernelCaps,
    pub precision: crate::fused::PrecisionGuarantee,
    pub cost: CostFn,
    /// Diagnostic tag identifying the kernel's implementation source
    /// when multiple alternatives register at the same `(op, dtypes,
    /// backend)` key. Used to distinguish e.g. AOCL's matmul from
    /// MKL's matmul from the portable scalar matmul — all three live
    /// at `(MatMul, [F32×3], Cpu)` as siblings.
    ///
    /// `""` is the default for kernels that don't need to distinguish.
    /// Conventional values: `"portable-cpu"` for fuel-cpu-backend
    /// scalar/gemm kernels, `"aocl"` for AOCL-BLAS, `"mkl"` for oneMKL,
    /// `"cublas"` / `"cutlass"` for CUDA variants. The string is
    /// diagnostic-only — never used for dispatch routing (the kernel
    /// function pointer is the dispatch identifier).
    pub kernel_source: &'static str,
    /// Whether this kernel's FKC accept-contract is the **GENERIC**
    /// (fully-permissive strided) one — admissible anywhere because it
    /// imposes no structure tightness (FKC §4.12). This is the
    /// structural-**miss** fallback bit: when the winning contract at a
    /// dispatch cell is generic, a structure-specialized kernel would have
    /// fit but none is registered, and the Baracuda miss telemetry records
    /// the demand.
    ///
    /// Computed **once at registration** by the FKC importer from the
    /// kernel's retained `ResolvedLayout` set
    /// ([`crate::fkc::is_generic_contract`]) and stamped here, so
    /// the live dispatch pick site reads a precomputed bool rather than
    /// re-deriving genericity from the lossy single-bool [`KernelCaps`].
    ///
    /// **Hand-written registrations default `false`.** Every non-FKC
    /// `register*` path routes through
    /// [`KernelBindingTable::register_full_with_source`], which sets this
    /// `false`: genericity is a claim about being the FKC generic-fallback
    /// contract in Baracuda's structure-key specialization space, and a
    /// hand-written kernel makes no such claim. This holds even for a
    /// hand-written strided kernel (`register_with_caps(strided_input=true)`):
    /// strided-capability is not the same as being the generic fallback, and
    /// conflating them would fabricate a miss demand signal for cells that are
    /// not part of Baracuda's matrix. Only the FKC importer, which computed
    /// genericity from the declared contract, ever sets this `true`.
    pub is_generic: bool,
    /// Stable per-implementation-version hash — the fifth axis of the FKC
    /// kernel-identity basis tuple (FKC §4.11), pinning the exact kernel build
    /// so a persisted plan / telemetry record re-resolves to it. `0`
    /// ([`crate::fused::KernelRevisionHash::UNTRACKED`]) for hand-written
    /// registrations (which carry no revision); the FKC importer threads the
    /// contract-computed revision from `ResolvedPrimitive.revision` here. Read
    /// by the Baracuda dispatch-telemetry `ImplId` at the plan pick site so the
    /// `chosen`/`candidates` ids carry a real revision instead of a stamped `0`.
    pub kernel_revision_hash: u64,
    /// The contract's parsed cost-expression AST (FKC §4.4 / adoption plan
    /// §2.3), when the kernel came from an FKC-imported contract that
    /// declares one. `&'static` so the entry stays `Copy` (`CompiledCostExpr`
    /// derives `Debug`, which is all `&'static T` needs to stay `Copy` /
    /// `Clone` / `Debug` itself). `None` for every hand-written registration
    /// (`register`/`register_full*`) — those carry no declared cost AST, only
    /// a `CostFn` fn-pointer. Only the FKC importer, once it lowers a
    /// contract's cost block to a live `CostFn` (a later Group-2 task),
    /// stamps `Some(&CompiledCostExpr)` here.
    pub cost_expr: Option<&'static crate::fkc::CompiledCostExpr>,
}

/// The binding table's op key — generalized so **runtime-registered** fused
/// ops live in the SAME registry as static kernels (the 2026-07-08
/// decisions-log end-state; retires the transitional runtime-kernel sidecar).
///
/// `Static` is the compile-time `OpKind` family every existing registration
/// uses (a bare `OpKind` converts via `From`, so those call sites are
/// unchanged). `RuntimeFused` names a Tier-2 (JIT-synthesized / import-time)
/// fused op by its runtime `FusedOpId` (`>= RUNTIME_FUSED_BASE`) — the id IS
/// the op identity, so no fake `OpKind` ever enters the table.
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
pub enum BindingKey {
    /// A compile-time op family (every static kernel registration).
    Static(OpKind),
    /// A runtime-registered fused op, keyed by its runtime id.
    RuntimeFused(FusedOpId),
}

impl From<OpKind> for BindingKey {
    fn from(op: OpKind) -> Self {
        BindingKey::Static(op)
    }
}

impl BindingKey {
    /// The `Static` op kind, if this is a static-family key.
    pub fn static_op(self) -> Option<OpKind> {
        match self {
            BindingKey::Static(op) => Some(op),
            BindingKey::RuntimeFused(_) => None,
        }
    }
}

/// `Clone` is load-bearing, not a convenience: the process-wide table is
/// published as a copy-on-write snapshot (`RwLock<Arc<KernelBindingTable>>`
/// in `dispatch.rs`), so a writer clones the current table, mutates the
/// clone, and swaps the pointer. Readers therefore never hold a lock guard.
/// The clone is shallow-ish and cheap — [`BindingEntry`] is `Copy` and the
/// keys are `Copy`/`SmallVec`, so it is a `HashMap` bulk copy of POD, on a
/// write-rare path (backend init + runtime-fusion adoption).
#[derive(Default, Clone)]
pub struct KernelBindingTable {
    bindings: HashMap<(BindingKey, KernelDTypes, BackendId), SmallVec<[BindingEntry; 2]>>,
}

/// The outcome of [`KernelBindingTable::unique_runtime_fused`] — three
/// distinct facts, deliberately not collapsed into an `Option`.
#[derive(Debug)]
pub enum RuntimeFusedLookup<'a> {
    /// No runtime-fused binding for this `(fid, backend)`.
    NotBound,
    /// Exactly one dtype tuple is registered; this is its first alternative.
    Unique(&'a [DType], &'a BindingEntry),
    /// Several dtype tuples share this id. **No kernel is returned**, because
    /// any choice among them would be arbitrary — and one of them may have a
    /// different operand arity than the caller expects.
    Ambiguous { signatures: Vec<Vec<DType>> },
}

impl KernelBindingTable {
    pub fn new() -> Self {
        Self {
            bindings: HashMap::new(),
        }
    }

    /// Register a dispatch wrapper for `(op, dtypes, backend)` with the
    /// default (all-false) capabilities. `dtypes` is the per-operand
    /// dtype list — inputs in order, then outputs.
    ///
    /// Phase 7.6 step 9a: multiple distinct kernels may register
    /// against the same `(op, dtypes, backend)` key — they become
    /// sibling alternatives at one decision point. Registration is
    /// **append-only and never panics**; the **same** `KernelRef`
    /// function pointer registered twice is detected by
    /// [`Self::finalize`] (run once after bulk registration) and
    /// returned as a `Result`, not raised as an inline panic.
    ///
    /// PrecisionGuarantee defaults to
    /// [`PrecisionGuarantee::UNAUDITED`](crate::fused::PrecisionGuarantee::UNAUDITED).
    /// Step-7b convention: the always-built backend
    /// (fuel-cpu-backend) runs [`Self::fill_unset_cpu_precision`] at
    /// the end of its bulk registration pass to upgrade every UNAUDITED
    /// CPU entry to `PRIMITIVE_DETERMINISTIC_CPU` (bit-stable per
    /// hardware). Kernels with weaker guarantees should call
    /// [`Self::register_with_precision`] explicitly *before* the fill
    /// pass to opt out of the default.
    pub fn register(
        &mut self,
        op: impl Into<BindingKey>,
        dtypes: &[DType],
        backend: BackendId,
        kernel: KernelRef,
    ) {
        self.register_full(
            op,
            dtypes,
            backend,
            kernel,
            KernelCaps::empty(),
            crate::fused::PrecisionGuarantee::UNAUDITED,
            unknown_cost,
        );
    }

    /// Register a dispatch wrapper with explicit capability flags.
    /// Used by binding sites that opt into kernel-side broadcast or
    /// other non-default behavior — the executor consults the caps to
    /// decide whether to auto-Contiguize inputs before kernel call.
    pub fn register_with_caps(
        &mut self,
        op: impl Into<BindingKey>,
        dtypes: &[DType],
        backend: BackendId,
        kernel: KernelRef,
        caps: KernelCaps,
    ) {
        self.register_full(
            op,
            dtypes,
            backend,
            kernel,
            caps,
            crate::fused::PrecisionGuarantee::UNAUDITED,
            unknown_cost,
        );
    }

    /// Register with an explicit [`crate::fused::PrecisionGuarantee`].
    /// Use this when a kernel has a precision claim that differs from
    /// the bulk-fill default (e.g., a multi-threaded reduction with
    /// non-deterministic accumulation order — bit_stable_on_same_hardware
    /// must be false).
    pub fn register_with_precision(
        &mut self,
        op: impl Into<BindingKey>,
        dtypes: &[DType],
        backend: BackendId,
        kernel: KernelRef,
        precision: crate::fused::PrecisionGuarantee,
    ) {
        self.register_full(
            op,
            dtypes,
            backend,
            kernel,
            KernelCaps::empty(),
            precision,
            unknown_cost,
        );
    }

    /// Backwards-compatible alias for full-form registration without
    /// an explicit cost function. New code should prefer
    /// [`Self::register_full`].
    pub fn register_with_caps_and_precision(
        &mut self,
        op: impl Into<BindingKey>,
        dtypes: &[DType],
        backend: BackendId,
        kernel: KernelRef,
        caps: KernelCaps,
        precision: crate::fused::PrecisionGuarantee,
    ) {
        self.register_full(op, dtypes, backend, kernel, caps, precision, unknown_cost);
    }

    /// Phase 7.6 step 8: full-form register with caps + precision +
    /// cost. Cost defaults to [`unknown_cost`] in the other
    /// signatures; sites with non-default cost claims must use this
    /// form (and not be overwritten by the bulk
    /// [`Self::fill_unset_cpu_cost`] pass, which only touches
    /// entries whose cost is still `unknown_cost`).
    ///
    /// Phase 7.6 step 9a: appends to the alternative set for `(op,
    /// dtypes, backend)`. Distinct `KernelRef` function pointers
    /// compose as siblings. Append-only — a duplicate `KernelRef` is
    /// detected by [`Self::finalize`] after bulk registration, not by
    /// an inline panic (see [`Self::register`]).
    pub fn register_full(
        &mut self,
        op: impl Into<BindingKey>,
        dtypes: &[DType],
        backend: BackendId,
        kernel: KernelRef,
        caps: KernelCaps,
        precision: crate::fused::PrecisionGuarantee,
        cost: CostFn,
    ) {
        self.register_full_with_source(op, dtypes, backend, kernel, caps, precision, cost, "");
    }

    /// Phase: backend-extensions refactor (2026-06-07). Same as
    /// [`Self::register_full`] but tags the binding with a
    /// `kernel_source` identifier used to distinguish kernels that
    /// register at the same `(op, dtypes, backend)` key from
    /// different implementation sources (AOCL vs MKL vs portable
    /// CPU, etc.). The tag is diagnostic-only — the picker still
    /// distinguishes alternatives by function-pointer identity.
    pub fn register_full_with_source(
        &mut self,
        op: impl Into<BindingKey>,
        dtypes: &[DType],
        backend: BackendId,
        kernel: KernelRef,
        caps: KernelCaps,
        precision: crate::fused::PrecisionGuarantee,
        cost: CostFn,
        kernel_source: &'static str,
    ) {
        // Hand-written registrations make no FKC generic-fallback claim —
        // see [`BindingEntry::is_generic`]. The FKC importer uses
        // [`Self::register_full_with_source_generic`] to stamp the computed
        // bit. Hand-written kernels carry no revision, so the revision axis is
        // `UNTRACKED` (`0`); only the FKC importer threads a real revision.
        self.register_full_with_source_generic(
            op,
            dtypes,
            backend,
            kernel,
            caps,
            precision,
            cost,
            kernel_source,
            false,
            crate::fused::KernelRevisionHash::UNTRACKED.0,
            None,
        );
    }

    /// Full-form register with the FKC structural-fallback `is_generic` bit
    /// (see [`BindingEntry::is_generic`]). The FKC importer computes the bit
    /// once from a kernel's retained `ResolvedLayout` set
    /// ([`crate::fkc::is_generic_contract`]) and stamps it here;
    /// every hand-written path routes through
    /// [`Self::register_full_with_source`], which passes `false`.
    ///
    /// Otherwise identical to [`Self::register_full_with_source`]:
    /// append-only, never-panic; the same `KernelRef` on one key is caught by
    /// [`Self::finalize`], not an inline panic.
    #[allow(clippy::too_many_arguments)]
    pub fn register_full_with_source_generic(
        &mut self,
        op: impl Into<BindingKey>,
        dtypes: &[DType],
        backend: BackendId,
        kernel: KernelRef,
        caps: KernelCaps,
        precision: crate::fused::PrecisionGuarantee,
        cost: CostFn,
        kernel_source: &'static str,
        is_generic: bool,
        kernel_revision_hash: u64,
        cost_expr: Option<&'static crate::fkc::CompiledCostExpr>,
    ) {
        let key = (op.into(), SmallVec::from_slice(dtypes), backend);
        let entry = BindingEntry {
            kernel,
            caps,
            precision,
            cost,
            kernel_source,
            is_generic,
            kernel_revision_hash,
            cost_expr,
        };
        // Append-only (never-panic): distinct `KernelRef`s compose as
        // sibling alternatives at one decision point; the *same*
        // function pointer registered twice is NOT rejected inline.
        // [`Self::finalize`] detects duplicates after bulk registration
        // and returns a `Result`, so the static init boundary can
        // fail-fast (`.expect()`) while a dynamic importer surfaces the
        // error — honoring never-panic on the import path.
        self.bindings.entry(key).or_default().push(entry);
    }

    /// Validate the fully-assembled table: within the alternative set
    /// for any one `(op, dtypes, backend)` key, each `KernelRef`
    /// function pointer must appear **at most once**. Distinct
    /// alternatives compose as siblings; the same pointer twice is a
    /// registration bug (a copy-paste in a hand-written table, or a
    /// contract importer wiring one kernel onto two keys that collapse).
    ///
    /// `register*` is append-only and never panics; this single pass —
    /// run once after bulk registration — surfaces a duplicate as a
    /// typed `Result`. The static init boundary
    /// ([`crate::dispatch::global_bindings`]) `.expect()`s it (fail-fast
    /// at startup for hand-written tables); the FKC contract importer
    /// will call it with `?`. This is the never-panic replacement for
    /// the former inline duplicate `panic!`.
    pub fn finalize(&self) -> Result<()> {
        for ((op, dtypes, backend), alts) in self.bindings.iter() {
            for i in 0..alts.len() {
                let ptr_i = alts[i].kernel as *const () as usize;
                for entry_j in &alts[i + 1..] {
                    if entry_j.kernel as *const () as usize == ptr_i {
                        return Err(Error::Msg(format!(
                            "KernelBindingTable: duplicate KernelRef registered \
                             for (op={op:?}, dtypes={dtypes:?}, backend={backend:?}). \
                             The same function pointer was registered twice; \
                             distinct alternatives at one decision point must be \
                             distinct `fn` items."
                        )));
                    }
                }
            }
        }
        Ok(())
    }

    /// Phase 7.6 step 7b: upgrade every UNAUDITED-precision CPU
    /// registration to the supplied `default`. Convention is to call
    /// this at the *end* of bulk registration, so any
    /// `register_with_precision(...)` calls that explicitly claimed a
    /// non-default value are preserved. The architecture-target
    /// shape would be precision-per-call-site, but for the ~335 CPU
    /// primitive registrations that all share the deterministic
    /// F32-accumulator property, a fill pass keeps the call sites
    /// concise without sacrificing the architectural commitment
    /// (every entry ends up with an explicit, non-UNAUDITED precision
    /// claim before lookup is ever exercised).
    ///
    /// Only entries with `backend == BackendId::Cpu` and the current
    /// `PrecisionGuarantee::UNAUDITED` sentinel are touched. Non-CPU
    /// backends register their own precision claims; non-UNAUDITED
    /// entries are preserved.
    ///
    /// Step 9a: applies to **every alternative** registered under a
    /// CPU key, not just the first — so a key with N siblings all
    /// starting at UNAUDITED ends with N entries upgraded to `default`.
    pub fn fill_unset_cpu_precision(&mut self, default: crate::fused::PrecisionGuarantee) {
        for ((_, _, backend), alts) in self.bindings.iter_mut() {
            if *backend != BackendId::Cpu {
                continue;
            }
            for entry in alts.iter_mut() {
                let p = &mut entry.precision;
                // Upgrade entries that claim nothing — EXCEPT an audited
                // "no static bound applies" conclusion.
                //
                // GAP-095: the old predicate was purely structural ("all four
                // value fields at defaults"), and the comment here asserted
                // that "a real weaker claim would set at least one of them."
                // That is false for `PrecisionGuarantee::none(reason)`, which
                // is field-identical to UNAUDITED and sets none of them. The
                // fill therefore overwrote an explicit audit conclusion with
                // `bit_stable_on_same_hardware: true` — asserting on the
                // kernel's behalf exactly what its audit had declined to
                // claim. FKC-4.8-0001 sanctions this fill for blocks left
                // absent or `audited: false`; it does not sanction that.
                //
                // Note what must KEEP being upgraded: a contract claim that
                // the V-FKC-9 ledger gate downgraded collapses to the
                // UNAUDITED sentinel exactly, so it stays eligible here. That
                // is load-bearing — every one of the CPU table's bit-stable
                // entries arrives through this path (GAP-077), so excluding
                // them would break the FKC-4.8-0001 coverage commitment
                // outright.
                if p.makes_no_static_claim() && !p.is_audited_no_bound() {
                    *p = default;
                }
            }
        }
    }

    /// Phase 7.6 step 8: upgrade every still-`unknown_cost` CPU
    /// registration to the cost function returned by `dispatcher` for
    /// its OpKind. The dispatcher's contract: given an `OpKind`,
    /// return the appropriate cost-family function — typically
    /// [`crate::cost::default_cost_for_op_kind`] in production code,
    /// but tests may pass their own dispatcher to exercise specific
    /// cost shapes.
    ///
    /// Convention is to call this at the *end* of bulk registration,
    /// so any `register_full(...)` calls that explicitly claimed a
    /// non-default cost are preserved. The function-pointer equality
    /// check identifies UNKNOWN entries — any caller that wants a
    /// "weaker" claim with non-zero values must use a distinct
    /// function pointer (which won't compare equal to `unknown_cost`
    /// and thus won't be overwritten).
    ///
    /// Step 9a: applies to **every alternative** with a CPU backend —
    /// each sibling entry at a key starting at `unknown_cost` is
    /// upgraded independently.
    pub fn fill_unset_cpu_cost(&mut self, dispatcher: fn(OpKind) -> CostFn) {
        self.fill_unset_cost_for_backend(BackendId::Cpu, dispatcher);
    }

    /// Per-backend variant of [`Self::fill_unset_cpu_cost`]. Walks
    /// every alternative whose backend matches `backend` and replaces
    /// `unknown_cost` entries with the dispatcher's choice for the
    /// op. Used by [`crate::vulkan_dispatch::register_vulkan_kernels`]
    /// to bulk-fill cost functions after per-kernel precision
    /// registration; future MKL/AOCL/Metal backends register through
    /// this same helper.
    ///
    /// Backend-specific cost adjustments (e.g. higher
    /// `kernel_overhead_ns` for Vulkan command-buffer submission vs
    /// CPU's nested-loop call overhead) are a Layer-2 refinement —
    /// today's `default_cost_for_op_kind` returns the same CPU-flavored
    /// functions regardless of backend, which over-estimates Vulkan
    /// throughput on small tensors. Either pass a Vulkan-flavored
    /// dispatcher to this method, or rely on the empirical calibration
    /// framework's per-(op, dtype, backend) refinement.
    pub fn fill_unset_cost_for_backend(
        &mut self,
        backend: BackendId,
        dispatcher: fn(OpKind) -> CostFn,
    ) {
        let sentinel = unknown_cost as *const () as usize;
        for ((key, _, this_backend), alts) in self.bindings.iter_mut() {
            if *this_backend != backend {
                continue;
            }
            // Runtime-fused entries keep the sentinel: their cost derives from
            // the op's recipe (keyed by the runtime id), not an OpKind table.
            let BindingKey::Static(op) = key else {
                continue;
            };
            for entry in alts.iter_mut() {
                if (entry.cost as usize) == sentinel {
                    entry.cost = dispatcher(*op);
                }
            }
        }
    }

    /// Look up the [`crate::fused::PrecisionGuarantee`] for a
    /// registered `(op, dtypes, backend)` triple. Returns the **first
    /// alternative**'s precision; multi-impl callers wanting all
    /// alternatives use [`Self::lookup_alternatives`]. Returns
    /// `PrecisionGuarantee::UNAUDITED` if the binding is missing —
    /// callers (notably the step-7b lint) treat that as "no claim"
    /// rather than panicking.
    pub fn lookup_precision(
        &self,
        op: impl Into<BindingKey>,
        dtypes: &[DType],
        backend: BackendId,
    ) -> crate::fused::PrecisionGuarantee {
        let key = (op.into(), SmallVec::from_slice(dtypes), backend);
        self.bindings
            .get(&key)
            .and_then(|alts| alts.first())
            .map(|e| e.precision)
            .unwrap_or(crate::fused::PrecisionGuarantee::UNAUDITED)
    }

    /// Look up the [`CostFn`] for a registered `(op, dtypes, backend)`
    /// triple. Returns the **first alternative**'s cost; multi-impl
    /// callers wanting all alternatives use
    /// [`Self::lookup_alternatives`]. Returns [`unknown_cost`] for
    /// missing bindings — the step-8 lint treats that as "no cost
    /// claim."
    pub fn lookup_cost(
        &self,
        op: impl Into<BindingKey>,
        dtypes: &[DType],
        backend: BackendId,
    ) -> CostFn {
        let key = (op.into(), SmallVec::from_slice(dtypes), backend);
        self.bindings
            .get(&key)
            .and_then(|alts| alts.first())
            .map(|e| e.cost)
            .unwrap_or(unknown_cost)
    }

    /// Iterate `(op, dtypes, backend, precision)` over every
    /// registered alternative — one tuple per `BindingEntry`, not per
    /// key. Used by the step-7b coverage lint, which groups by
    /// OpKind and checks the bit-stable commitment per group; the
    /// grouping handles N-alternatives-per-key naturally.
    pub fn iter_precision(
        &self,
    ) -> impl Iterator<
        Item = (
            OpKind,
            &[DType],
            BackendId,
            crate::fused::PrecisionGuarantee,
        ),
    > {
        self.bindings
            .iter()
            .filter_map(|((key, dtypes, backend), alts)| {
                // Static-kernel audit view: runtime-fused entries are outside the
                // per-OpKind commitments these iterators exist to check.
                key.static_op().map(|op| (op, dtypes, backend, alts))
            })
            .flat_map(|(op, dtypes, backend, alts)| {
                alts.iter()
                    .map(move |e| (op, dtypes.as_slice(), *backend, e.precision))
            })
    }

    /// Iterate `(op, dtypes, backend, &entry)` over every registered
    /// alternative — one tuple per `BindingEntry`, mirroring
    /// [`Self::iter_precision`] but yielding the FULL entry (by reference)
    /// instead of projecting out just the `precision` field. Used by the
    /// FKC empirical verification harness (Task 4.4+), which needs the
    /// whole `BindingEntry` (the `kernel` fn pointer to invoke, `caps`,
    /// `kernel_revision_hash` for ledger keys, ...), not just one field.
    /// Runtime-fused/JIT entries are filtered out (`static_op()` returns
    /// `None` for them) — same "static-kernel audit view" as
    /// `iter_precision`/`iter_cost`.
    pub fn iter_entries(
        &self,
    ) -> impl Iterator<Item = (OpKind, &[DType], BackendId, &BindingEntry)> {
        self.bindings
            .iter()
            .filter_map(|((key, dtypes, backend), alts)| {
                key.static_op().map(|op| (op, dtypes, backend, alts))
            })
            .flat_map(|(op, dtypes, backend, alts)| {
                alts.iter()
                    .map(move |e| (op, dtypes.as_slice(), *backend, e))
            })
    }

    /// Iterate `(op, dtypes, backend, cost_fn)` over every registered
    /// alternative — one tuple per `BindingEntry`. Used by the step-8
    /// coverage lint to check the "every primitive op has a
    /// non-default cost function" commitment per OpKind.
    pub fn iter_cost(&self) -> impl Iterator<Item = (OpKind, &[DType], BackendId, CostFn)> {
        self.bindings
            .iter()
            .filter_map(|((key, dtypes, backend), alts)| {
                key.static_op().map(|op| (op, dtypes, backend, alts))
            })
            .flat_map(|(op, dtypes, backend, alts)| {
                alts.iter()
                    .map(move |e| (op, dtypes.as_slice(), *backend, e.cost))
            })
    }

    /// Iterate every registered `(op, dtypes, backend)` decision-point
    /// key — one item per key, regardless of how many alternatives
    /// share it. Consumed by `SystemTopology::build_at` to derive
    /// which backends have kernels registered + which (op, dtype)
    /// pairs each backend covers.
    pub fn iter_keys(&self) -> impl Iterator<Item = (OpKind, &[DType], BackendId)> {
        self.bindings.keys().filter_map(|(key, dtypes, backend)| {
            key.static_op().map(|op| (op, dtypes.as_slice(), *backend))
        })
    }

    /// Whether ANY runtime-fused entry is bound for `(fid, backend)`,
    /// regardless of dtype tuple — the coarse capability-gate query (the
    /// optimizer's fuse/don't-fuse predicate is id-level; the dtype-precise
    /// check is the dispatch-time [`Self::lookup_with_caps`] itself).
    pub fn has_runtime_fused(&self, fid: FusedOpId, backend: BackendId) -> bool {
        self.bindings
            .keys()
            .any(|(k, _, b)| *b == backend && matches!(k, BindingKey::RuntimeFused(f) if *f == fid))
    }

    /// Resolve `(fid, backend)` to a runtime-fused binding, **refusing to pick
    /// when the answer is ambiguous**.
    ///
    /// A [`FusedOpId`] names a RECIPE, not a signature: `register_runtime_fused`
    /// deduplicates on the region's base-map hash, so one id can legitimately
    /// carry several dtype tuples — `[F32, F32]` and `[F32, F32, F32]` are
    /// different arities under the same id. The pre-GAP-213 `first_runtime_fused` scanned
    /// for `(fid, backend)` and **binds `dtypes` without testing it**, so it
    /// answers such a question by `HashMap::iter()` order: a coin flip per
    /// process, and capable of returning a kernel of the wrong operand arity.
    ///
    /// The three outcomes are kept as three values rather than folded into an
    /// `Option`, because *not bound* and *ambiguous* are different facts with
    /// different fixes, and an `Option` would make them indistinguishable to the
    /// caller — the same collapse that let GAP-001 read as a version regression
    /// for eleven days.
    ///
    /// **Scope: this is the diagnostic/test resolution path.** Production
    /// dispatch never came through here — it uses [`Self::lookup_with_caps`],
    /// which takes the dtype tuple and does an exact keyed `get`, so it can
    /// neither be ambiguous nor return a wrong arity. See GAP-213.
    pub fn unique_runtime_fused(
        &self,
        fid: FusedOpId,
        backend: BackendId,
    ) -> RuntimeFusedLookup<'_> {
        let mut matches = self.bindings.iter().filter(|((k, _, b), _)| {
            *b == backend && matches!(k, BindingKey::RuntimeFused(f) if *f == fid)
        });
        let Some(((_, dtypes, _), alts)) = matches.next() else {
            return RuntimeFusedLookup::NotBound;
        };
        if matches.next().is_some() {
            let mut signatures: Vec<Vec<DType>> = self
                .bindings
                .iter()
                .filter(|((k, _, b), _)| {
                    *b == backend && matches!(k, BindingKey::RuntimeFused(f) if *f == fid)
                })
                .map(|((_, d, _), _)| d.to_vec())
                .collect();
            // HashMap order is per-process, so the report would otherwise vary run to
            // run. `DType` is not `Ord`, so key on arity then the debug spelling —
            // a presentation order only; nothing depends on which comes first.
            signatures.sort_by_key(|d| (d.len(), format!("{d:?}")));
            return RuntimeFusedLookup::Ambiguous { signatures };
        }
        match alts.first() {
            Some(e) => RuntimeFusedLookup::Unique(dtypes.as_slice(), e),
            None => RuntimeFusedLookup::NotBound,
        }
    }

    /// **TEST-ONLY.** Drop every runtime-fused entry (static entries retained).
    /// The runtime-fused reset story: the metadata sidecar's Vec length is the
    /// id allocator, so clearing it MUST clear these bindings in the same
    /// breath or a reused id resolves a stale kernel
    /// (`runtime_fused_kernels::clear_runtime_fused_for_tests` does both).
    #[doc(hidden)]
    pub fn remove_runtime_fused_for_tests(&mut self) {
        self.bindings
            .retain(|(k, _, _), _| matches!(k, BindingKey::Static(_)));
    }

    /// Look up the wrapper for `(op, dtypes, backend)`. Returns the
    /// **first alternative**'s [`KernelRef`]; for capability-aware
    /// lookup use [`Self::lookup_with_caps`]; for the full alternative
    /// set use [`Self::lookup_alternatives`].
    pub fn lookup(
        &self,
        op: impl Into<BindingKey>,
        dtypes: &[DType],
        backend: BackendId,
    ) -> Result<KernelRef> {
        self.lookup_with_caps(op, dtypes, backend).map(|(k, _)| k)
    }

    /// Capability-aware lookup. Returns the first **dense-safe** alternative
    /// — the first sibling whose caps do NOT set
    /// [`KernelCaps::requires_broadcast`] — paired with its registered
    /// [`KernelCaps`]. A baked-broadcast sibling (`broadcast_stride0:
    /// required`) is SKIPPED, so a dense operand is never routed into a
    /// kernel that reads `in[0]`; the exclusion is order-independent (path
    /// 1a). For every existing binding — none set `requires_broadcast` —
    /// this is byte-identical to returning the first-registered alternative.
    /// Surfaces [`Error::NoBackendForOp`] on a missing binding OR when every
    /// sibling is broadcast-baked (production-correct: a surfaced miss, never
    /// a wrong read, no panic).
    pub fn lookup_with_caps(
        &self,
        op: impl Into<BindingKey>,
        dtypes: &[DType],
        backend: BackendId,
    ) -> Result<(KernelRef, KernelCaps)> {
        let key = (op.into(), SmallVec::from_slice(dtypes), backend);
        self.bindings
            .get(&key)
            .and_then(|alts| alts.iter().find(|e| !e.caps.requires_broadcast))
            .map(|e| (e.kernel, e.caps))
            .ok_or_else(|| {
                let (available_backends, supported_combinations) = self.coverage_diagnostic();
                Error::NoBackendForOp {
                    op: key.0.static_op().unwrap_or(OpKind::RuntimeFused),
                    dtypes: dtypes.to_vec(),
                    available_backends,
                    supported_combinations,
                }
                .bt()
            })
    }

    /// Build the truthful `(available_backends, supported_combinations)`
    /// diagnostic pair for a [`Error::NoBackendForOp`] miss: the DISTINCT
    /// registered backends, and the full static `(backend, op, dtypes)` coverage
    /// (the useful "what exists" context). Shared by
    /// [`lookup_with_caps`](Self::lookup_with_caps) AND the plan-time miss
    /// (`plan::missing_binding_error`) so BOTH surface the same real context. The
    /// plan-time path formerly hardcoded `[]` here, which falsely reads as "no
    /// backend is registered" and misdirects diagnosis — a mixed-precision
    /// `(op, dtypes)` gap is not an empty registry (observed in the wild:
    /// `[F32, BF16, F32]` matmul reported as "available backends: []").
    pub fn coverage_diagnostic(&self) -> (Vec<BackendId>, Vec<CoverageBinding>) {
        let available_backends: Vec<BackendId> = self
            .bindings
            .keys()
            .map(|(_, _, b)| *b)
            .collect::<std::collections::HashSet<_>>()
            .into_iter()
            .collect();
        let supported_combinations: Vec<(BackendId, OpKind, Vec<DType>)> = self
            .bindings
            .keys()
            .filter_map(|(k, d, b)| k.static_op().map(|o| (*b, o, d.to_vec())))
            .collect();
        (available_backends, supported_combinations)
    }

    /// Does **this specific backend** serve `(op, dtypes)` with a usable
    /// (non-broadcast-baked) binding?
    ///
    /// Mirrors [`lookup_with_caps`](Self::lookup_with_caps)'s success condition
    /// exactly — same key, same `!requires_broadcast` admission — but skips
    /// constructing the miss diagnostic, so it is cheap enough to call in a
    /// per-node loop (`coverage_diagnostic` walks the entire table).
    ///
    /// Ask this once per *candidate* backend. There is deliberately no
    /// "does any backend anywhere serve it?" sibling: a predicate that ignores
    /// the key's backend component cannot answer a dispatch question, because a
    /// node executes on one backend rather than on the union of them. Such a
    /// sibling did exist (`any_backend_supports`) and it caused a real defect —
    /// the dtype-reconciliation pass consulted it, so a backend a node could
    /// never run on suppressed the `Op::Cast` that node needed and planning
    /// dead-ended at `NoBackendForOp`. See `crate::plan::candidate_backends_for`.
    pub fn backend_supports(
        &self,
        op: impl Into<BindingKey>,
        dtypes: &[DType],
        backend: BackendId,
    ) -> bool {
        let key = (op.into(), SmallVec::from_slice(dtypes), backend);
        self.bindings
            .get(&key)
            .is_some_and(|alts| alts.iter().any(|e| !e.caps.requires_broadcast))
    }

    /// Phase 7.6 step 9a: return the full set of registered
    /// alternatives at the `(op, dtypes, backend)` decision point.
    /// Order is registration order — append-on-register is the step-9a
    /// contract. The empty slice means "no kernel registered at this
    /// decision point."
    ///
    /// The route picker (step 9b) consumes this to rank alternatives
    /// by cost + precision at plan time. Single-impl callers continue
    /// to use [`Self::lookup_with_caps`] (which returns the first
    /// alternative — the route picker's job is the longer-term home
    /// for selection logic).
    pub fn lookup_alternatives(
        &self,
        op: impl Into<BindingKey>,
        dtypes: &[DType],
        backend: BackendId,
    ) -> &[BindingEntry] {
        let key = (op.into(), SmallVec::from_slice(dtypes), backend);
        self.bindings
            .get(&key)
            .map(|alts| alts.as_slice())
            .unwrap_or(&[])
    }

    /// Total number of registered alternatives across all keys (not
    /// just unique keys). Step 9a: a key with N siblings counts N.
    pub fn len(&self) -> usize {
        self.bindings.values().map(|alts| alts.len()).sum()
    }

    /// Empty binding table.
    pub fn is_empty(&self) -> bool {
        self.bindings.is_empty()
    }

    /// Total number of unique `(op, dtypes, backend)` keys. Step 9a:
    /// distinct from [`Self::len`], which counts alternatives.
    pub fn key_count(&self) -> usize {
        self.bindings.len()
    }
}

impl std::fmt::Debug for KernelBindingTable {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("KernelBindingTable")
            .field("keys", &self.bindings.len())
            .field("total_alternatives", &self.len())
            .finish_non_exhaustive()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use fuel_ir::DType;

    /// Smoke: KernelRef can be constructed and stored.
    #[test]
    fn kernel_ref_stores_function_pointer() {
        fn dummy_kernel(
            _inputs: &[Arc<RwLock<Storage>>],
            _outputs: &mut [Arc<RwLock<Storage>>],
            _layouts: &[Layout],
            _params: &OpParams,
        ) -> Result<()> {
            Ok(())
        }
        let k: KernelRef = dummy_kernel;
        // Function pointer is Copy + Clone.
        let _k2 = k;
        let _k3: KernelRef = k;
    }

    /// Smoke: OpParams variants construct cleanly and Debug-format.
    #[test]
    fn op_params_variants_construct() {
        let _ = OpParams::None;
        let _ = OpParams::Reduce {
            dims: vec![0, 1],
            keepdim: false,
        };
        let _ = OpParams::Matmul {
            lhs_batch_dims: vec![],
            rhs_batch_dims: vec![],
            m: 4,
            n: 8,
            k: 16,
            m_compute: MatmulM::All,
        };
        let _ = OpParams::Slice {
            dim: 0,
            start: 0,
            end: 10,
            step: 1,
        };
        let _ = OpParams::Cast;
        let _ = OpParams::Affine { mul: 2.0, add: 1.0 };
        // Debug format compiles.
        let _ = format!("{:?}", OpParams::None);
    }

    /// Smoke: a hand-constructed kernel that allocates the output
    /// Storage outside this crate would be:
    ///   1. allocate output via fuel_memory::alloc_cpu_zeroed
    ///   2. wrap inputs as &[Arc<RwLock<Storage>>]
    ///   3. call the kernel
    ///
    /// Phase B5 lands the first such migration; B1 just type-checks
    /// the surface.
    #[test]
    fn kernel_ref_can_be_called() {
        fn copy_kernel(
            inputs: &[Arc<RwLock<Storage>>],
            outputs: &mut [Arc<RwLock<Storage>>],
            _layouts: &[Layout],
            _params: &OpParams,
        ) -> Result<()> {
            // Simplest real kernel: copy bytes from inputs[0] to outputs[0].
            let in_arc = &inputs[0];
            let out_arc = &outputs[0];
            let in_guard = in_arc.read().unwrap();
            let mut out_guard = out_arc.write().unwrap();
            let in_bytes = match &in_guard.inner {
                fuel_memory::BackendStorage::Cpu(s) => s.bytes(),
                #[allow(unreachable_patterns)]
                _ => return Ok(()),
            };
            if let fuel_memory::BackendStorage::Cpu(s) = &mut out_guard.inner {
                s.bytes_mut().copy_from_slice(in_bytes);
            }
            Ok(())
        }

        let input = fuel_memory::from_slice_cpu(&[1.0_f32, 2.0, 3.0, 4.0]);
        let output = fuel_memory::alloc_cpu_zeroed(DType::F32, 4).unwrap();
        let inputs = vec![Arc::new(RwLock::new(input))];
        let mut outputs = vec![Arc::new(RwLock::new(output))];

        let k: KernelRef = copy_kernel;
        k(&inputs, &mut outputs, &[], &OpParams::None).unwrap();

        // Output bytes match input.
        let out_guard = outputs[0].read().unwrap();
        if let fuel_memory::BackendStorage::Cpu(s) = &out_guard.inner {
            let typed: &[f32] = s.as_slice().unwrap();
            assert_eq!(typed, &[1.0, 2.0, 3.0, 4.0]);
        }
    }

    fn ok_kernel(
        _inputs: &[Arc<RwLock<Storage>>],
        _outputs: &mut [Arc<RwLock<Storage>>],
        _layouts: &[Layout],
        _params: &OpParams,
    ) -> Result<()> {
        Ok(())
    }

    fn ok_kernel_alt(
        _inputs: &[Arc<RwLock<Storage>>],
        _outputs: &mut [Arc<RwLock<Storage>>],
        _layouts: &[Layout],
        _params: &OpParams,
    ) -> Result<()> {
        Ok(())
    }

    /// Phase 7.6 step 9a: two distinct `KernelRef`s register against
    /// the same `(op, dtypes, backend)` key as sibling alternatives.
    /// `lookup_with_caps` returns the first; `lookup_alternatives`
    /// returns both in registration order.
    #[test]
    fn step_9a_two_alternatives_append_then_first_wins_on_legacy_lookup() {
        use fuel_ir::probe::BackendId;
        let mut table = KernelBindingTable::new();
        let dts = [DType::BF16, DType::BF16, DType::BF16];
        table.register(OpKind::MatMul, &dts, BackendId::Cuda, ok_kernel);
        table.register(OpKind::MatMul, &dts, BackendId::Cuda, ok_kernel_alt);

        // Legacy single-impl lookup returns the first-registered.
        let (k1, _caps) = table
            .lookup_with_caps(OpKind::MatMul, &dts, BackendId::Cuda)
            .unwrap();
        assert_eq!(k1 as *const () as usize, ok_kernel as *const () as usize);

        // Multi-impl lookup returns both in registration order.
        let alts = table.lookup_alternatives(OpKind::MatMul, &dts, BackendId::Cuda);
        assert_eq!(alts.len(), 2);
        assert_eq!(
            alts[0].kernel as *const () as usize,
            ok_kernel as *const () as usize
        );
        assert_eq!(
            alts[1].kernel as *const () as usize,
            ok_kernel_alt as *const () as usize
        );

        // Step-9a accounting helpers stay consistent.
        assert_eq!(table.key_count(), 1, "one decision point");
        assert_eq!(table.len(), 2, "two alternatives total");
    }

    /// FKC gap-closure Task 2.1: hand-written registrations (the
    /// `register`/`register_full*` family that never touches the FKC
    /// importer) carry no declared cost-expression AST — `cost_expr`
    /// defaults to `None`. Only the FKC importer (a later Group-2 task)
    /// stamps a real `Some(&CompiledCostExpr)`.
    #[test]
    fn cost_expr_field_defaults_none_for_hand_written_registration() {
        use fuel_ir::probe::BackendId;
        let mut table = KernelBindingTable::new();
        let dts = [DType::F32, DType::F32, DType::F32];
        table.register(OpKind::AddElementwise, &dts, BackendId::Cpu, ok_kernel);
        let alts = table.lookup_alternatives(OpKind::AddElementwise, &dts, BackendId::Cpu);
        assert_eq!(alts.len(), 1);
        assert!(
            alts[0].cost_expr.is_none(),
            "hand-written registrations carry no declared cost AST"
        );
    }

    /// Task 4.4: `iter_entries` mirrors `iter_precision`/`iter_cost` —
    /// yields one `(op, dtypes, backend, &entry)` tuple per registered
    /// alternative (both siblings at a shared key show up), and filters
    /// out `RuntimeFused` keys just like the precision/cost audit
    /// iterators (a "static-kernel audit view").
    #[test]
    fn iter_entries_yields_full_entries_and_filters_runtime_fused() {
        use fuel_ir::probe::BackendId;
        let mut table = KernelBindingTable::new();
        let dts = [DType::F32, DType::F32, DType::F32];
        table.register(OpKind::AddElementwise, &dts, BackendId::Cpu, ok_kernel);
        table.register(OpKind::AddElementwise, &dts, BackendId::Cpu, ok_kernel_alt);
        // A runtime-fused key — must be excluded from `iter_entries`.
        table.register(
            BindingKey::RuntimeFused(FusedOpId(FusedOpId::RUNTIME_FUSED_BASE)),
            &dts,
            BackendId::Cpu,
            ok_kernel,
        );

        let entries: Vec<_> = table.iter_entries().collect();
        assert_eq!(
            entries.len(),
            2,
            "both AddElementwise alternatives, RuntimeFused excluded"
        );
        for (op, dtypes, backend, entry) in &entries {
            assert_eq!(*op, OpKind::AddElementwise);
            assert_eq!(*dtypes, dts.as_slice());
            assert_eq!(*backend, BackendId::Cpu);
            // Full entry is reachable, not just one projected field.
            let _ = entry.caps;
            let _ = entry.kernel_revision_hash;
        }
        let fns: Vec<usize> = entries
            .iter()
            .map(|(_, _, _, e)| e.kernel as *const () as usize)
            .collect();
        assert!(fns.contains(&(ok_kernel as *const () as usize)));
        assert!(fns.contains(&(ok_kernel_alt as *const () as usize)));
    }

    /// Never-panic (option B): registering the same `KernelRef` twice
    /// against one key is append-only — it does NOT panic at
    /// registration. `finalize()` detects the duplicate and returns
    /// `Err`. (Replaces the former `#[should_panic]` guard.)
    #[test]
    /// FKC-CLAUSE: FKC-10.10-0001 — the boundary itself: registration stays
    /// infallible and `finalize()` RETURNS the duplicate as an error rather than
    /// panicking. This is why the clause does not require a `Result` signature on
    /// `register_full_with_source`.
    fn step_9a_duplicate_kernel_ref_detected_by_finalize_not_panic() {
        use fuel_ir::probe::BackendId;
        let mut table = KernelBindingTable::new();
        let dts = [DType::F32, DType::F32, DType::F32];
        table.register(OpKind::MatMul, &dts, BackendId::Cpu, ok_kernel);
        table.register(OpKind::MatMul, &dts, BackendId::Cpu, ok_kernel); // dup — must NOT panic
        let err = table
            .finalize()
            .expect_err("finalize must report the duplicate KernelRef");
        assert!(
            format!("{err}").contains("duplicate KernelRef"),
            "unexpected error: {err}"
        );
    }

    /// Distinct sibling kernels under one key finalize cleanly.
    #[test]
    fn finalize_ok_for_distinct_sibling_kernels() {
        use fuel_ir::probe::BackendId;
        let mut table = KernelBindingTable::new();
        let dts = [DType::F32, DType::F32, DType::F32];
        table.register(OpKind::MatMul, &dts, BackendId::Cpu, ok_kernel);
        table.register(OpKind::MatMul, &dts, BackendId::Cpu, ok_kernel_alt);
        table
            .finalize()
            .expect("distinct sibling kernels must finalize OK");
    }

    /// Path 1a: a `broadcast_stride0: required` (baked-broadcast) kernel
    /// reads its operand as stride-0; a dense operand routed there
    /// miscomputes (reads `in[0]` repeatedly). The realize pick MUST skip
    /// such a sibling when a generic one exists, and — the load-bearing
    /// part — that exclusion must NOT rest on the generic being registered
    /// first. Register the broadcast-baked one FIRST (adversarial: `first()`
    /// would return it) and assert the pick returns the GENERIC sibling.
    #[test]
    fn lookup_with_caps_skips_broadcast_baked_sibling_regardless_of_order() {
        use fuel_ir::probe::BackendId;
        let mut table = KernelBindingTable::new();
        let dts = [DType::F32, DType::F32, DType::F32];
        // Baked-broadcast sibling registered FIRST.
        table.register_with_caps(
            OpKind::AddElementwise,
            &dts,
            BackendId::Cpu,
            ok_kernel,
            KernelCaps {
                strided_input: true,
                requires_broadcast: true,
            },
        );
        // Generic dense-safe sibling registered SECOND.
        table.register_with_caps(
            OpKind::AddElementwise,
            &dts,
            BackendId::Cpu,
            ok_kernel_alt,
            KernelCaps {
                strided_input: true,
                requires_broadcast: false,
            },
        );

        let (picked, caps) = table
            .lookup_with_caps(OpKind::AddElementwise, &dts, BackendId::Cpu)
            .expect("a dense-safe sibling exists");
        assert!(
            !caps.requires_broadcast,
            "the realize pick must never be a broadcast-baked kernel",
        );
        assert_eq!(
            picked as *const () as usize, ok_kernel_alt as *const () as usize,
            "lookup_with_caps must skip the broadcast-baked sibling and return the generic \
             one regardless of registration order",
        );
        // Both siblings coexist — the exclusion is at the pick, not at
        // registration (the baked-broadcast contract is still PRESENT).
        let alts = table.lookup_alternatives(OpKind::AddElementwise, &dts, BackendId::Cpu);
        assert_eq!(
            alts.len(),
            2,
            "both siblings registered; neither is dropped"
        );
    }

    /// The corner that keeps the exclusion honest: if EVERY sibling is
    /// broadcast-baked (no dense-safe fallback), the pick is a surfaced
    /// miss (`NoBackendForOp`), NOT a silent wrong dense read.
    #[test]
    fn lookup_with_caps_misses_when_only_broadcast_baked_siblings_exist() {
        use fuel_ir::probe::BackendId;
        let mut table = KernelBindingTable::new();
        let dts = [DType::F32, DType::F32, DType::F32];
        table.register_with_caps(
            OpKind::AddElementwise,
            &dts,
            BackendId::Cpu,
            ok_kernel,
            KernelCaps {
                strided_input: true,
                requires_broadcast: true,
            },
        );
        assert!(
            table
                .lookup_with_caps(OpKind::AddElementwise, &dts, BackendId::Cpu)
                .is_err(),
            "no dense-safe sibling ⇒ honest miss, never a wrong read",
        );
    }

    /// Phase 7.6 step 9a: `fill_unset_cpu_precision` upgrades every
    /// alternative under a CPU key, not just the first.
    #[test]
    fn step_9a_fill_passes_touch_every_alternative() {
        use crate::fused::PrecisionGuarantee;
        use fuel_ir::probe::BackendId;
        let mut table = KernelBindingTable::new();
        let dts = [DType::F32, DType::F32, DType::F32];
        table.register(OpKind::AddElementwise, &dts, BackendId::Cpu, ok_kernel);
        table.register(OpKind::AddElementwise, &dts, BackendId::Cpu, ok_kernel_alt);

        // Both start at UNAUDITED.
        for e in table.lookup_alternatives(OpKind::AddElementwise, &dts, BackendId::Cpu) {
            assert!(!e.precision.bit_stable_on_same_hardware);
        }

        let bit_stable = PrecisionGuarantee {
            bit_stable_on_same_hardware: true,
            max_ulp: Some(0),
            max_relative: None,
            max_absolute: None,
            notes: "test default",
        };
        table.fill_unset_cpu_precision(bit_stable);

        // Both got upgraded.
        let alts = table.lookup_alternatives(OpKind::AddElementwise, &dts, BackendId::Cpu);
        assert_eq!(alts.len(), 2);
        for e in alts {
            assert!(e.precision.bit_stable_on_same_hardware);
            assert_eq!(e.precision.max_ulp, Some(0));
        }
    }

    /// GAP-095: the CPU bulk fill must NOT overwrite an audited
    /// "no static bound applies" conclusion.
    ///
    /// `PrecisionGuarantee::none(reason)` and `PrecisionGuarantee::UNAUDITED`
    /// carry identical *values* — all four claim fields at their defaults —
    /// and differ only in `notes`. The fill's predicate was purely
    /// structural, so it conflated them and rewrote an explicit audit
    /// conclusion into `PRIMITIVE_DETERMINISTIC_CPU`: it asserted
    /// `bit_stable_on_same_hardware: true` on a kernel whose audit had
    /// specifically declined to claim it.
    ///
    /// FKC-4.8-0001 sanctions the fill for precision blocks left *absent* or
    /// `audited: false`. It does not sanction overwriting an `audited: true`
    /// + `none(reason)` conclusion.
    ///
    /// **Both directions asserted.** A test that only checked that
    /// `none(reason)` survives would also pass with the fill disabled
    /// outright, so the UNAUDITED sibling must still be upgraded in the same
    /// run — otherwise the test proves nothing about discrimination.
    #[test]
    fn gap_095_fill_preserves_audited_none_but_still_upgrades_unaudited() {
        use crate::fused::PrecisionGuarantee;
        use fuel_ir::probe::BackendId;
        let mut table = KernelBindingTable::new();
        let dts = [DType::F32, DType::F32, DType::F32];

        const REASON: &str = "audited: accumulation order is scheduler-dependent; \
                              no static bound applies";
        // The audited conclusion — must survive untouched.
        table.register_with_precision(
            OpKind::MatMul,
            &dts,
            BackendId::Cpu,
            ok_kernel,
            PrecisionGuarantee::none(REASON),
        );
        // The never-audited placeholder — must still be upgraded.
        table.register(OpKind::AddElementwise, &dts, BackendId::Cpu, ok_kernel);

        table.fill_unset_cpu_precision(PrecisionGuarantee::PRIMITIVE_DETERMINISTIC_CPU);

        let audited = table
            .lookup_alternatives(OpKind::MatMul, &dts, BackendId::Cpu)
            .first()
            .expect("audited-none entry registered")
            .precision;
        assert!(
            !audited.bit_stable_on_same_hardware,
            "the fill asserted bit-stability on a kernel whose audit \
             explicitly concluded that no static bound applies",
        );
        assert_eq!(
            audited.notes, REASON,
            "the audit reason was overwritten — the reasoning that lived at \
             the registration site is gone",
        );

        // Opposite direction: the fill must still do its job.
        let placeholder = table
            .lookup_alternatives(OpKind::AddElementwise, &dts, BackendId::Cpu)
            .first()
            .expect("unaudited entry registered")
            .precision;
        assert!(
            placeholder.bit_stable_on_same_hardware,
            "control FAILED: the UNAUDITED placeholder was not upgraded, so \
             this test would pass with the fill disabled entirely and proves \
             nothing about the discrimination it is meant to check",
        );
    }
}