hrx/

loom.rs

//! In-process Loom compilation with reusable indexed modules and verified disk artifacts.
#[allow(
    non_camel_case_types,
    non_snake_case,
    non_upper_case_globals,
    dead_code,
    missing_docs,
    unsafe_op_in_unsafe_fn,
    clippy::all
)]
mod ffi;
mod native;
mod report;
mod source;
use crate::{
    Error, Result,
    bundle::{self, Lock},
};
pub use report::{
    CompileReport, EntryChange, EntryResources, Expansion, ReportGuidance, ReportSuggestion,
    WaitCounts, WaitReason, WaitReasonChange,
};
use serde::{Deserialize, Serialize};
pub use source::{CxxSource, CxxStandard, Source};
use std::{
    collections::{BTreeMap, HashMap, HashSet},
    fs,
    num::NonZeroUsize,
    path::{Path, PathBuf},
    sync::{Arc, Mutex},
};

/// How seriously the compiler meant a diagnostic. Ordered, so callers can
/// filter with a comparison rather than a magic number: a spill remark and a
/// type error arrive in the same list, and only one of them is worth a reader.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Severity {
    /// A remark: informational, including backend resource notes.
    Note,
    /// A warning: compilation continued.
    Warning,
    /// An error: compilation of this specialization failed.
    Error,
}
impl From<u32> for Severity {
    /// Unknown native severities are treated as errors, so a future compiler
    /// cannot make a diagnostic disappear from a caller's error filter.
    fn from(native: u32) -> Self {
        match native {
            0 => Self::Note,
            1 => Self::Warning,
            _ => Self::Error,
        }
    }
}

/// A compiler diagnostic copied out of native result storage.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Diagnostic {
    /// How seriously the compiler meant this diagnostic.
    pub severity: Severity,
    /// Stable compiler diagnostic code.
    pub code: String,
    /// Rendered diagnostic message.
    pub message: String,
    /// Full compiler-rendered diagnostic, including source context when available.
    pub formatted_text: String,
    /// Named values in diagnostic-definition order.
    pub parameters: Vec<DiagnosticParameter>,
    /// Diagnostic source identifier, when supplied by the compiler.
    #[serde(default)]
    pub source: String,
    /// One-based source line, or zero when unavailable.
    pub line: u32,
    /// One-based source column, or zero when unavailable.
    pub column: u32,
    /// Labeled locations providing context, in compiler order.
    #[serde(default)]
    pub related_locations: Vec<RelatedLocation>,
    /// Additional locations omitted by the compiler's diagnostic limit.
    #[serde(default)]
    pub related_location_omitted_count: usize,
}

/// One compiler-rendered diagnostic argument.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct DiagnosticParameter {
    /// Stable name from the compiler diagnostic definition.
    pub name: String,
    /// Canonical rendered value.
    pub value: String,
}

/// A source location related to a compiler diagnostic, owned by Rust.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct RelatedLocation {
    /// Relationship to the primary diagnostic, such as a previous use.
    pub label: String,
    /// Source identifier, even when the compiler did not retain source text.
    pub source: String,
    /// One-based starting line, or zero when unavailable.
    pub line: u32,
    /// One-based starting column, or zero when unavailable.
    pub column: u32,
    /// One-based ending line, or zero when unavailable.
    pub end_line: u32,
    /// One-based ending column, or zero when unavailable.
    pub end_column: u32,
}
/// AMDGPU workgroup scheduling domain. Explicit modes require GFX11/GFX12.
#[derive(Clone, Copy, Debug, Default, Hash, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ProcessorMode {
    /// Preserve the compiler's default for the selected target.
    #[default]
    Default,
    /// Schedule each workgroup within a single compute unit (GFX11/GFX12).
    ComputeUnit,
    /// Schedule across a workgroup processor (GFX11/GFX12).
    WorkgroupProcessor,
}

/// Assertion classes inserted at the target's semantic assertion boundary.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct SanitizerChecks {
    /// Memory access assertions (requires the target's access-shadow runtime).
    pub access: bool,
    /// SSA value assertions.
    pub value: bool,
    /// Operation assertions.
    pub operation: bool,
    /// Data-race observations (requires the target's race-shadow runtime).
    pub race: bool,
}
impl SanitizerChecks {
    pub(super) fn bits(self) -> u64 {
        u64::from(self.access)
            | (u64::from(self.value) << 1)
            | (u64::from(self.operation) << 2)
            | (u64::from(self.race) << 3)
    }
}
/// Target behavior on a failed sanitizer assertion.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum SanitizerReporting {
    /// Native target's structured report path.
    #[default]
    Default,
    /// Direct fatal trap without structured reports.
    Trap,
    /// Structured report without a direct trap.
    ReportOnly,
}
/// Native compiler instrumentation. Runtime-required globals are described by
/// the artifact manifest; compiling does not supply a sanitizer runtime.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct SanitizerOptions {
    /// Independent classes of assertions.
    pub checks: SanitizerChecks,
    /// Behavior of failed assertions.
    pub reporting: SanitizerReporting,
}

/// Compiler target and limits for concurrent workspaces and cached modules.
#[derive(Clone, Debug)]
pub struct CompilerOptions {
    /// Maximum number of simultaneous compilations through this compiler.
    pub workers: NonZeroUsize,
    /// Architecture used by the Loom target profile and artifact cache.
    pub target: crate::Target,
    /// Maximum retained source modules; zero disables module caching.
    /// Eviction chooses an arbitrary entry, not the least recently used module.
    pub module_cache_capacity: usize,
    /// Workgroup scheduling policy for the selected AMDGPU target.
    pub processor_mode: ProcessorMode,
    /// Instrumentation included in pipeline, shared compiler, and artifact identities.
    pub sanitizer: SanitizerOptions,
}
impl Default for CompilerOptions {
    fn default() -> Self {
        Self {
            target: crate::Target::default(),
            module_cache_capacity: 64,
            processor_mode: ProcessorMode::Default,
            sanitizer: SanitizerOptions::default(),
            workers: NonZeroUsize::new(
                std::thread::available_parallelism()
                    .map_or(1, NonZeroUsize::get)
                    .min(4),
            )
            .unwrap(),
        }
    }
}
impl CompilerOptions {
    /// Default compiler limits for an explicit device target.
    #[must_use]
    pub fn for_target(target: &crate::Target) -> Self {
        Self {
            target: target.clone(),
            ..Self::default()
        }
    }
    /// Default compiler limits targeting the device behind `stream`.
    #[must_use]
    pub fn for_stream(stream: &crate::Stream) -> Self {
        Self::for_target(stream.target())
    }
}
struct Inner {
    // Indexes are released before prepared compiler state when the session ends.
    modules: Mutex<HashMap<String, Arc<ModuleData>>>,
    native: native::Prepared,
    target: crate::Target,
    workers: NonZeroUsize,
    module_cache_capacity: usize,
    processor_mode: ProcessorMode,
    sanitizer: SanitizerOptions,
    path: PathBuf,
    identity: String,
}

#[derive(Hash, PartialEq, Eq)]
struct SharedCompilerKey {
    path: PathBuf,
    target: String,
    workers: NonZeroUsize,
    module_cache_capacity: usize,
    processor_mode: ProcessorMode,
    sanitizer: SanitizerOptions,
}
/// A pinned compiler library and reusable native state, shared by cheap clones.
#[derive(Clone)]
pub struct Compiler(Arc<Inner>);
impl std::fmt::Debug for Compiler {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Compiler")
            .field("library", &self.0.path)
            .field("identity", &self.0.identity)
            .finish_non_exhaustive()
    }
}
impl Compiler {
    /// Select an explicit library, HRX_LOOM_LIBRARY, or the pinned native bundle.
    /// This loads CPU compiler code but never opens a GPU. Library mappings stay
    /// resident until process exit; contexts and workspaces remain session-owned.
    /// To upgrade a loaded library, select a new path or restart the process.
    pub fn resolve(library: Option<&Path>) -> Result<Self> {
        Self::with_options(library, CompilerOptions::default())
    }
    /// Resolve the shared compiler configured for a stream's actual device target.
    pub fn for_stream(library: Option<&Path>, stream: &crate::Stream) -> Result<Self> {
        Self::shared(library, CompilerOptions::for_stream(stream))
    }
    /// Resolve the shared compiler configured for an explicit device target.
    pub fn for_target(library: Option<&Path>, target: &crate::Target) -> Result<Self> {
        Self::shared(library, CompilerOptions::for_target(target))
    }
    /// The compiler for this library, target and limits, resolved once per process.
    ///
    /// Resolving is not cheap: it canonicalizes the library path and digests
    /// the whole shared object, twice. A consumer that asks for a compiler per
    /// kernel pays that per kernel, so every consumer memoized it themselves.
    ///
    /// The cache includes all compiler options. `HRX_LOOM_LIBRARY` is resolved
    /// on each call rather than frozen at the first.
    pub fn shared(library: Option<&Path>, options: CompilerOptions) -> Result<Self> {
        static RESOLVED: Mutex<Option<HashMap<SharedCompilerKey, Compiler>>> = Mutex::new(None);
        let key = SharedCompilerKey {
            path: Self::resolved_library(library)?,
            target: options.target.as_str().to_owned(),
            workers: options.workers,
            module_cache_capacity: options.module_cache_capacity,
            processor_mode: options.processor_mode,
            sanitizer: options.sanitizer,
        };
        let mut cache = RESOLVED
            .lock()
            .map_err(|_| Error::Message("compiler cache poisoned".into()))?;
        let cache = cache.get_or_insert_with(HashMap::new);
        if let Some(compiler) = cache.get(&key) {
            return Ok(compiler.clone());
        }
        let compiler = Self::with_options(Some(&key.path), options)?;
        cache.insert(key, compiler.clone());
        Ok(compiler)
    }

    /// The library a given override, `HRX_LOOM_LIBRARY` or the bundle selects.
    fn resolved_library(library: Option<&Path>) -> Result<PathBuf> {
        match library
            .map(Path::to_path_buf)
            .or_else(|| std::env::var_os("HRX_LOOM_LIBRARY").map(PathBuf::from))
        {
            Some(p) => fs::canonicalize(&p)
                .map_err(|e| Error::from(e).context(format!("compiler library {}", p.display()))),
            None => Ok(fs::canonicalize(bundle::resolve()?.join("libloomc.so"))?),
        }
    }

    /// Resolve a compiler with a bounded number of exclusive workspaces.
    pub fn with_options(library: Option<&Path>, options: CompilerOptions) -> Result<Self> {
        let path = Self::resolved_library(library)?;
        let identity = bundle::file_digest(&path)?;
        let native = native::Prepared::open(
            &path,
            &identity,
            options.workers.get(),
            &options.target,
            options.processor_mode,
            options.sanitizer,
        )?;
        if bundle::file_digest(&path)? != identity {
            return Err(Error::Message(
                "compiler library changed while loading".into(),
            ));
        }
        Ok(Self(Arc::new(Inner {
            modules: Mutex::new(HashMap::new()),
            native,
            target: options.target,
            workers: options.workers,
            module_cache_capacity: options.module_cache_capacity,
            processor_mode: options.processor_mode,
            sanitizer: options.sanitizer,
            path,
            identity,
        })))
    }
    /// Maximum simultaneous compilations, and the workspace pool's bound.
    pub fn workers(&self) -> NonZeroUsize {
        self.0.workers
    }

    /// Compile several specializations, up to [`CompilerOptions::workers`] at a
    /// time, and return their outcomes in request order.
    ///
    /// [`Module::compile`] blocks, so a single-threaded caller never reaches the
    /// workspace pool's bound. This is the entry point that does: it owns the
    /// thread budget rather than leaving every consumer to rebuild the same pool.
    /// Each request is independent; one failure does not cancel the others.
    pub fn compile_all(&self, requests: &[(&Module, &Specialization)]) -> Vec<Result<Artifact>> {
        let limit = self.0.workers.get().min(requests.len());
        if limit <= 1 {
            return requests
                .iter()
                .map(|(module, spec)| module.compile(spec))
                .collect();
        }
        let next = std::sync::atomic::AtomicUsize::new(0);
        let slots: Vec<Mutex<Option<Result<Artifact>>>> =
            requests.iter().map(|_| Mutex::new(None)).collect();
        std::thread::scope(|scope| {
            for _ in 0..limit {
                scope.spawn(|| {
                    loop {
                        let index = next.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                        let Some((module, spec)) = requests.get(index) else {
                            return;
                        };
                        let outcome = module.compile(spec);
                        *slots[index].lock().unwrap_or_else(|e| e.into_inner()) = Some(outcome);
                    }
                });
            }
        });
        slots
            .into_iter()
            .map(|slot| {
                slot.into_inner()
                    .unwrap_or_else(|e| e.into_inner())
                    .expect("every request slot is filled before the scope ends")
            })
            .collect()
    }

    /// Architecture selected for every compilation by this compiler.
    pub fn target(&self) -> &crate::Target {
        &self.0.target
    }
    /// Content identity of the loaded compiler library.
    pub fn identity(&self) -> &str {
        &self.0.identity
    }
    /// Selected compiler library path.
    pub fn library_path(&self) -> &Path {
        &self.0.path
    }
    /// Retain a source module, sharing its parsed index across specializations.
    /// Parsing is deferred until the first artifact-cache miss.
    pub fn module(&self, source: &str) -> Module {
        self.source_module(vec![Source::loom("kernel.loom", source)])
    }
    /// Link named Loom and C/C++ inputs in one compiler-owned source module.
    /// Source bytes, headers, macro order and import options all enter its identity.
    pub fn sources(&self, mut sources: Vec<Source>) -> Result<Module> {
        if sources.is_empty() {
            return Err(Error::Message(
                "a compilation module requires source inputs".into(),
            ));
        }
        let mut identifiers = std::collections::BTreeSet::new();
        for source in &mut sources {
            source.validate()?;
            let identifier = match source {
                Source::Cxx(unit) => {
                    unit.normalize()?;
                    &unit.identifier
                }
                Source::Loom { identifier, .. } => identifier,
            };
            if !identifiers.insert(identifier.clone()) {
                return Err(Error::Message("duplicate source identifier".into()));
            }
        }
        Ok(self.source_module(sources))
    }
    /// Import a translation unit; parsing is deferred until compilation.
    pub fn import_cxx(&self, source: CxxSource) -> Result<Module> {
        self.sources(vec![Source::Cxx(source)])
    }
    fn source_module(&self, sources: Vec<Source>) -> Module {
        let digest = bundle::digest(
            &serde_json::to_vec(&sources).expect("source inputs contain only JSON-safe values"),
        );
        let mut modules = self.0.modules.lock().unwrap_or_else(|e| e.into_inner());
        let data = if let Some(data) = modules.get(&digest) {
            data.clone()
        } else {
            let data = Arc::new(ModuleData {
                sources,
                digest: digest.clone(),
                index: Mutex::new(None),
            });
            if self.0.module_cache_capacity != 0 {
                if modules.len() >= self.0.module_cache_capacity {
                    let victim = modules.keys().next().cloned().unwrap();
                    modules.remove(&victim);
                }
                modules.insert(digest, data.clone());
            }
            data
        };
        Module {
            compiler: self.clone(),
            data,
        }
    }
    /// Release idle workspace blocks and cached module references. Existing
    /// Module handles remain valid; active compilations are unaffected.
    pub fn trim(&self) {
        self.0
            .modules
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .clear();
        self.0.native.trim();
    }
}
/// Render the diagnostic a caller should act on, not the whole cascade.
///
/// A single rejected construct makes the rest of its block unparseable, so the
/// compiler reports one real error followed by a run of consequences. Leading
/// with all of them buries the cause; every diagnostic stays available through
/// [`Error::Compile::diagnostics`].
fn summarize(diagnostics: &[Diagnostic]) -> String {
    let errors = || diagnostics.iter().filter(|d| d.severity >= Severity::Error);
    let Some(first) = errors().next() else {
        return diagnostics
            .iter()
            .map(|d| d.message.as_str())
            .collect::<Vec<_>>()
            .join("\n");
    };
    let mut message = if first.line == 0 {
        first.message.clone()
    } else {
        format!("{}:{}: {}", first.line, first.column, first.message)
    };
    if let Some(hint) = hint(&first.message) {
        message.push_str("\nhint: ");
        message.push_str(hint);
    }
    for related in &first.related_locations {
        message.push_str(&format!(
            "\n{}:{}:{}: {}",
            related.source, related.line, related.column, related.label
        ));
    }
    if first.related_location_omitted_count != 0 {
        message.push_str(&format!(
            "\n({} additional related location(s) omitted)",
            first.related_location_omitted_count
        ));
    }
    let suppressed = errors().count() - 1;
    if suppressed > 0 {
        message.push_str(&format!(
            "\n({suppressed} later error(s) suppressed as cascade; see Error::Compile diagnostics)"
        ));
    }
    message
}

/// Diagnostics whose cause is a configuration mistake rather than a source bug.
fn hint(message: &str) -> Option<&'static str> {
    if message.contains("Low representation contract") && message.contains(".generic.") {
        return Some(
            "a generic target has no low-asm contract; name a bare architecture \
             such as gfx1151 in amdgpu.target<...> to use hand-written asm",
        );
    }
    None
}

#[cfg(test)]
mod diagnostic_tests {
    use super::{Diagnostic, Severity, summarize};
    fn diagnostic(severity: u32, line: u32, message: &str) -> Diagnostic {
        Diagnostic {
            severity: Severity::from(severity),
            code: String::new(),
            message: message.into(),
            formatted_text: message.into(),
            parameters: Vec::new(),
            source: String::new(),
            line,
            column: 1,
            related_locations: Vec::new(),
            related_location_omitted_count: 0,
        }
    }
    #[test]
    fn the_first_error_leads_and_its_cascade_is_counted_not_printed() {
        let mut all = vec![
            diagnostic(1, 3, "unused binding"),
            diagnostic(
                2,
                7,
                "unknown Low representation contract amdgpu.gfx11.generic.core",
            ),
        ];
        all.extend((0..12).map(|i| diagnostic(2, 8 + i, "expected expression")));
        let rendered = summarize(&all);
        assert!(rendered.starts_with("7:1: unknown Low representation contract"));
        assert!(rendered.contains("hint: a generic target has no low-asm contract"));
        assert!(rendered.contains("12 later error(s) suppressed"));
        assert!(!rendered.contains("expected expression"));
        // A cascade-free failure gains no suppression line and no hint.
        let single = summarize(&[diagnostic(2, 4, "type mismatch")]);
        assert_eq!(single, "4:1: type mismatch");
    }
}

struct ModuleData {
    sources: Vec<Source>,
    digest: String,
    index: Mutex<Option<(Arc<native::Index>, Vec<Diagnostic>)>>,
}
/// Immutable source plus compiler ownership. Clones share the frozen native index.
#[derive(Clone)]
pub struct Module {
    // Release the index before the last compiler/context owner.
    data: Arc<ModuleData>,
    compiler: Compiler,
}
/// Amount of compiler evidence collected alongside an executable.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ReportMode {
    /// Emit no report and avoid detail collection.
    #[default]
    None,
    /// Collect artifact facts and summary compiler analysis.
    Summary,
    /// Include detailed provenance, scheduling, and resource evidence.
    Details,
}

/// Compiler-formatted pass trace representation.
#[derive(Clone, Copy, Debug, Default)]
pub enum TraceFormat {
    /// Human-readable metadata followed by source.
    Text,
    /// One JSON object per pass event.
    #[default]
    JsonLines,
}
/// Bounded streaming diagnostics for a fresh compile invocation.
#[derive(Clone, Debug)]
pub struct TraceOptions {
    /// Serialization chosen by the compiler.
    pub format: TraceFormat,
    /// Maximum bytes forwarded to the writer; exceeding this aborts compilation.
    pub max_bytes: usize,
    /// Pass keys or compiler stage names to capture before execution.
    pub before: Vec<String>,
    /// Pass keys or compiler stage names to capture after execution.
    /// Both filter lists empty selects every after boundary.
    pub after: Vec<String>,
}
impl Default for TraceOptions {
    fn default() -> Self {
        Self {
            format: TraceFormat::JsonLines,
            max_bytes: 16 * 1024 * 1024,
            before: Vec::new(),
            after: Vec::new(),
        }
    }
}

/// One export and its exact configuration within a module.
#[derive(Clone, Debug, Default)]
pub struct Specialization {
    /// Export/root symbol to compile.
    symbol: String,
    /// Fully qualified configuration keys and Loom value spellings.
    config: BTreeMap<String, String>,
    /// Select compiler analysis collected beside the executable.
    report: ReportMode,
}
impl Specialization {
    /// Select an export with no configuration overrides.
    pub fn new(symbol: impl Into<String>) -> Self {
        Self {
            symbol: symbol.into(),
            ..Self::default()
        }
    }
    /// Set one fully-qualified Loom configuration value.
    #[must_use]
    pub fn with_config(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
        self.config.insert(key.into(), value.into());
        self
    }
    /// Set one fully-qualified Loom configuration value in place.
    pub fn set_config(&mut self, key: impl Into<String>, value: impl Into<String>) -> &mut Self {
        self.config.insert(key.into(), value.into());
        self
    }
    /// Select an export/root symbol in place.
    pub fn set_symbol(&mut self, symbol: impl Into<String>) -> &mut Self {
        self.symbol = symbol.into();
        self
    }
    /// Replace all fully-qualified Loom configuration values.
    pub fn replace_config(&mut self, config: BTreeMap<String, String>) -> &mut Self {
        self.config = config;
        self
    }
    /// Request or suppress the compiler resource report.
    #[must_use]
    pub fn with_report(mut self, report: ReportMode) -> Self {
        self.report = report;
        self
    }
    /// Request or suppress the compiler resource report in place.
    pub fn set_report(&mut self, report: ReportMode) -> &mut Self {
        self.report = report;
        self
    }
    /// Export/root symbol selected by this specialization.
    #[must_use]
    pub fn symbol(&self) -> &str {
        &self.symbol
    }
    /// Fully-qualified compiler configuration.
    #[must_use]
    pub fn configuration(&self) -> &BTreeMap<String, String> {
        &self.config
    }
    /// Compiler evidence requested by this specialization.
    #[must_use]
    pub fn report_mode(&self) -> ReportMode {
        self.report
    }
    fn validate(&self) -> Result<()> {
        let valid = |s: &str| {
            !s.is_empty()
                && s.bytes()
                    .all(|b| b.is_ascii_alphanumeric() || b"._-".contains(&b))
        };
        if !valid(&self.symbol) || self.config.keys().any(|k| !valid(k)) {
            return Err(Error::Message(
                "invalid Loom export or configuration key".into(),
            ));
        }
        Ok(())
    }
}
#[derive(Clone, Debug, Serialize, Deserialize)]
struct Record {
    key: String,
    compiler: String,
    target: String,
    #[serde(default)]
    processor_mode: ProcessorMode,
    symbol: String,
    sha256: String,
    diagnostics: Vec<Diagnostic>,
    report: Option<CompileReport>,
    manifest: Option<serde_json::Value>,
    launch_sha256: Option<String>,
}
/// Owned native executable bytes and their verified compilation metadata.
#[derive(Clone, Debug)]
pub struct Artifact {
    bytes: Arc<[u8]>,
    launch_bytes: Option<Arc<[u8]>>,
    launch_loader: native::LaunchLoader,
    path: PathBuf,
    record: Record,
}
impl Artifact {
    /// Native executable bytes, independent of compiler/result lifetimes.
    pub fn bytes(&self) -> &[u8] {
        &self.bytes
    }
    /// Compiler-owned description of exports, arguments, globals, and target facts.
    pub fn manifest(&self) -> Option<&serde_json::Value> {
        self.record.manifest.as_ref()
    }
    /// Serialized host launch program produced alongside these executable bytes.
    pub fn launch_bytes(&self) -> Option<&[u8]> {
        self.launch_bytes.as_deref()
    }
    /// Bind an exact executable export to an independent launch evaluator.
    /// The matching native evaluator library stays owned by the artifact.
    pub fn launch_program(&self, export: &str) -> Result<LaunchProgram> {
        self.launch_loader.load(
            self.launch_bytes.as_deref().ok_or_else(|| {
                Error::Unsupported("this target has no host launch program".into())
            })?,
            export,
        )
    }
    /// Durable verified cache artifact path.
    pub fn path(&self) -> &Path {
        &self.path
    }
    /// Compiler library content identity.
    pub fn compiler_identity(&self) -> &str {
        &self.record.compiler
    }
    /// Exact GPU target key.
    pub fn target(&self) -> &str {
        &self.record.target
    }
    /// Scheduling domain used to compile this artifact.
    pub fn processor_mode(&self) -> ProcessorMode {
        self.record.processor_mode
    }
    /// Export compiled into this artifact.
    pub fn symbol(&self) -> &str {
        &self.record.symbol
    }
    /// Diagnostics retained from successful preparation and compilation.
    pub fn diagnostics(&self) -> &[Diagnostic] {
        &self.record.diagnostics
    }
    /// Optional structured compiler evidence, in the pinned compiler's schema.
    pub fn report(&self) -> Option<&CompileReport> {
        self.record.report.as_ref()
    }
}
/// Concrete launch properties evaluated from compiler-authored workload arguments.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct LaunchConfig {
    /// Workgroup count along each axis.
    pub workgroup_count: [u32; 3],
    /// Invocations per workgroup along each axis.
    pub workgroup_size: [u32; 3],
    /// Cooperative cluster dimensions; [1, 1, 1] for an ordinary launch.
    pub workgroup_cluster_size: [u32; 3],
    /// Hardware subgroup width, or zero when the target does not fix it.
    pub subgroup_size: u32,
    /// Total workgroup-local storage, including statically allocated storage.
    pub workgroup_storage_bytes: u64,
}

/// One bound launch function with exclusively owned invocation scratch.
/// Create one per independently prepared execution slot.
pub struct LaunchProgram(native::LaunchProgram);
impl LaunchProgram {
    /// Evaluate positional raw scalar bits without compiling source or looking up names.
    /// Evaluation may grow scratch; perform it while preparing a dispatch.
    pub fn evaluate(&mut self, workload_argument_bits: &[u64]) -> Result<LaunchConfig> {
        self.0.evaluate(workload_argument_bits)
    }
}

struct Compiled {
    bytes: Arc<[u8]>,
    diagnostics: Vec<Diagnostic>,
    report: Option<serde_json::Value>,
    manifest: Option<serde_json::Value>,
    launch_bytes: Option<Arc<[u8]>>,
}
impl Module {
    /// Identity of the exact source bytes.
    pub fn identity(&self) -> &str {
        &self.data.digest
    }
    /// Cache identity including compiler, pipeline contract, source and specialization.
    pub fn key(&self, spec: &Specialization) -> Result<String> {
        spec.validate()?;
        let base = bundle::digest(&serde_json::to_vec(&(
            "loomc-v5-sanitizers",
            self.compiler.0.sanitizer,
            self.compiler.identity(),
            self.identity(),
            &spec.symbol,
            self.compiler.0.target.as_str(),
            &spec.config,
            spec.report,
        ))?);
        // Retain existing default artifacts; explicit execution modes own a
        // separate namespace even if their output bytes happen to match.
        if self.compiler.0.processor_mode == ProcessorMode::Default {
            Ok(base)
        } else {
            Ok(bundle::digest(&serde_json::to_vec(&(
                "loomc-processor-mode-v1",
                base,
                self.compiler.0.processor_mode,
            ))?))
        }
    }
    /// Compile in process or return verified cached bytes. Publication is atomic
    /// and serialized per key across threads and processes; failures are retryable.
    pub fn compile(&self, spec: &Specialization) -> Result<Artifact> {
        self.compile_impl(spec, None)
    }
    /// Recompile and stream compiler-owned pass traces, even on a cache hit.
    /// A writer error, panic, or exhausted byte limit aborts compilation and
    /// leaves any existing cached artifact intact. The writer is not retained.
    pub fn compile_traced(
        &self,
        spec: &Specialization,
        options: &TraceOptions,
        writer: &mut dyn std::io::Write,
    ) -> Result<Artifact> {
        self.compile_impl(spec, Some((options, writer)))
    }
    fn compile_impl(
        &self,
        spec: &Specialization,
        trace: Option<(&TraceOptions, &mut dyn std::io::Write)>,
    ) -> Result<Artifact> {
        let key = self.key(spec)?;
        let cache = &bundle::kernel_cache()?;
        let dir = cache.join(&key);
        if trace.is_none()
            && let Some(a) = cached(&dir, &key, &self.compiler.0.native)
        {
            return Ok(a);
        }
        bundle::create_cache_dir(cache)?;
        let _lock = Lock::acquire(&cache.join(format!("{key}.lock")))?;
        if trace.is_none()
            && let Some(a) = cached(&dir, &key, &self.compiler.0.native)
        {
            return Ok(a);
        }
        let (index, mut diagnostics) = {
            let mut slot = self.data.index.lock().unwrap_or_else(|e| e.into_inner());
            if slot.is_none() {
                let (index, diagnostics) = self.compiler.0.native.index(&self.data.sources)?;
                *slot = Some((Arc::new(index), diagnostics));
            }
            slot.as_ref().unwrap().clone()
        };
        let compiled = self
            .compiler
            .0
            .native
            .compile(&index, spec, trace)
            .map_err(|e| {
                e.context(format!(
                    "compiling {} with {}",
                    spec.symbol,
                    self.compiler.library_path().display()
                ))
            })?;
        diagnostics.extend(compiled.diagnostics);
        let record = Record {
            key,
            compiler: self.compiler.identity().into(),
            target: self.compiler.0.target.as_str().into(),
            processor_mode: self.compiler.0.processor_mode,
            symbol: spec.symbol.clone(),
            sha256: bundle::digest(&compiled.bytes),
            launch_sha256: compiled.launch_bytes.as_deref().map(bundle::digest),
            manifest: compiled.manifest,
            diagnostics,
            report: compiled
                .report
                .map(|json| {
                    CompileReport::new(
                        self.compiler.identity().into(),
                        self.compiler.0.target.as_str().into(),
                        self.compiler.0.processor_mode,
                        json,
                    )
                })
                .transpose()?,
        };
        let staging = tempfile::tempdir_in(cache)?;
        let filename = self.compiler.target().artifact_filename();
        let output = staging.path().join(filename);
        fs::write(&output, &compiled.bytes)?;
        if let Some(bytes) = &compiled.launch_bytes {
            fs::write(staging.path().join("launch.loombc"), bytes)?;
            fs::File::open(staging.path().join("launch.loombc"))?.sync_all()?;
        }
        let metadata = serde_json::to_vec(&record)?;
        fs::write(staging.path().join("artifact.json"), &metadata)?;
        fs::write(
            staging.path().join("artifact.sha256"),
            bundle::digest(&metadata),
        )?;
        for file in [filename, "artifact.json", "artifact.sha256"] {
            fs::File::open(staging.path().join(file))?.sync_all()?;
        }
        fs::File::open(staging.path())?.sync_all()?;
        if dir.exists() {
            fs::remove_dir_all(&dir)?;
        }
        bundle::publish(staging, &dir)?;
        fs::File::open(cache)?.sync_all()?;
        Ok(Artifact {
            bytes: compiled.bytes,
            launch_bytes: compiled.launch_bytes,
            launch_loader: self.compiler.0.native.launch_loader(),
            path: dir.join(filename),
            record,
        })
    }
}
fn cached(dir: &Path, key: &str, native: &native::Prepared) -> Option<Artifact> {
    let metadata = fs::read(dir.join("artifact.json")).ok()?;
    if fs::read_to_string(dir.join("artifact.sha256")).ok()? != bundle::digest(&metadata) {
        return None;
    }
    let record: Record = serde_json::from_slice(&metadata).ok()?;
    if record.key != key {
        return None;
    }
    // `bundle::collect` evicts on access time. Reading the artifact below updates
    // it on a normal mount, but a `noatime` mount never would, so set it here:
    // utimensat is unaffected by the mount option that suppresses the implicit
    // update, and without this a busy cache would look idle to the collector.
    if let Ok(file) = fs::File::open(dir.join("artifact.json")) {
        let _ = file.set_times(fs::FileTimes::new().set_accessed(std::time::SystemTime::now()));
    }
    let target = crate::Target::new(&record.target).ok()?;
    let path = dir.join(target.artifact_filename());
    let bytes = fs::read(&path).ok()?;
    if bytes.is_empty() || bundle::digest(&bytes) != record.sha256 {
        return None;
    }
    let launch_bytes = if let Some(hash) = &record.launch_sha256 {
        let bytes = fs::read(dir.join("launch.loombc")).ok()?;
        if bytes.is_empty() || &bundle::digest(&bytes) != hash {
            return None;
        }
        Some(Arc::from(bytes))
    } else {
        None
    };
    Some(Artifact {
        bytes: bytes.into(),
        launch_bytes,
        launch_loader: native.launch_loader(),
        path,
        record,
    })
}

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

    #[test]
    #[ignore = "requires libloomc.so, no GPU"]
    fn shared_compilers_respect_all_limits_and_reuse_matching_options() -> Result<()> {
        let options = CompilerOptions {
            workers: NonZeroUsize::new(2).unwrap(),
            module_cache_capacity: 2,
            ..Default::default()
        };
        let first = Compiler::shared(None, options.clone())?;
        let same = Compiler::shared(None, options.clone())?;
        assert!(Arc::ptr_eq(&first.0, &same.0));
        let serial_options = CompilerOptions {
            workers: NonZeroUsize::new(1).unwrap(),
            ..options
        };
        let serial = Compiler::shared(None, serial_options.clone())?;
        assert_eq!(serial.workers().get(), 1);
        assert!(!Arc::ptr_eq(&first.0, &serial.0));
        let uncached = Compiler::shared(
            None,
            CompilerOptions {
                module_cache_capacity: 0,
                ..serial_options
            },
        )?;
        assert!(!Arc::ptr_eq(&serial.0, &uncached.0));
        serial.module("// retained");
        uncached.module("// not retained");
        assert!(!serial.0.modules.lock().unwrap().is_empty());
        assert!(uncached.0.modules.lock().unwrap().is_empty());
        Ok(())
    }

    #[test]
    #[ignore = "requires libloomc.so, no GPU"]
    fn processor_modes_have_distinct_compilers_and_artifacts() -> Result<()> {
        let default = Compiler::shared(None, CompilerOptions::default())?;
        let options = CompilerOptions {
            processor_mode: ProcessorMode::ComputeUnit,
            ..Default::default()
        };
        let cu = Compiler::shared(None, options.clone())?;
        let same = Compiler::shared(None, options)?;
        assert!(Arc::ptr_eq(&cu.0, &same.0));
        assert!(!Arc::ptr_eq(&default.0, &cu.0));
        let source = include_str!("../tests/kernels/euler.loom");
        let spec = euler_spec();
        let wgp = default.module(source).compile(&spec)?;
        let unit = cu.module(source).compile(&spec)?;
        assert_ne!(wgp.path(), unit.path());
        assert_eq!(unit.processor_mode(), ProcessorMode::ComputeUnit);
        let cached = cu.module(source).compile(&spec)?;
        assert_eq!(cached.processor_mode(), ProcessorMode::ComputeUnit);
        assert_eq!(unit.bytes(), cached.bytes());
        let differences: Vec<_> = wgp
            .bytes()
            .iter()
            .zip(unit.bytes())
            .filter(|(a, b)| a != b)
            .collect();
        assert_eq!(wgp.bytes().len(), unit.bytes().len());
        assert_eq!(differences.len(), 1);
        assert_eq!(*differences[0].0 ^ *differences[0].1, 0x20);
        Ok(())
    }

    fn euler_spec() -> Specialization {
        let mut spec = Specialization::new("krea2_euler");
        spec.set_config("krea2.euler.grid_x", "1");
        spec.set_config("krea2.euler.grid_y", "1");
        spec
    }

    #[test]
    #[ignore = "requires libloomc.so, no GPU"]
    fn reporting_preserves_shared_requests_and_can_be_replaced_from_a_callback() -> Result<()> {
        use std::sync::atomic::{AtomicUsize, Ordering};
        let kernels = Kernels::new(Compiler::resolve(None)?);
        let clone = kernels.clone();
        let pending =
            kernels.request(include_str!("../tests/kernels/euler.loom"), &euler_spec())?;
        let calls = Arc::new(AtomicUsize::new(0));
        let reported = calls.clone();
        let weak = Arc::downgrade(&kernels.0);
        let kernels = kernels.reporting(move |_| {
            reported.fetch_add(1, Ordering::SeqCst);
            // Replacing a reporter must not hold its lock while invoking it.
            // A weak reference avoids a callback/cache ownership cycle.
            let next = reported.clone();
            Kernels(weak.upgrade().unwrap()).reporting(move |_| {
                next.fetch_add(10, Ordering::SeqCst);
            });
        });
        assert!(Arc::ptr_eq(&kernels.0, &clone.0));
        assert!(Arc::ptr_eq(&kernels.0, &pending.cache));
        for cache in [&clone.0, &pending.cache] {
            let mut attempts = 0;
            let outcome = cache.build_with(1, |artifact| {
                attempts += 1;
                assert_eq!(artifact.record.key, pending.key);
                Err(Error::Message("injected load failure".into()))
            });
            assert_eq!(outcome.unwrap_err().to_string(), "injected load failure");
            assert_eq!(attempts, 1);
        }
        assert_eq!(calls.load(Ordering::SeqCst), 11);
        assert_eq!(kernels.0.locked()?.waiting.len(), 1);
        Ok(())
    }

    #[test]
    #[ignore = "requires libloomc.so, no GPU"]
    fn requests_during_build_are_deduplicated_and_new_requests_survive() -> Result<()> {
        let kernels = Kernels::new(Compiler::resolve(None)?);
        let source = include_str!("../tests/kernels/euler.loom");
        let spec = euler_spec();
        let pending = kernels.request(source, &spec)?;
        let mut other_spec = spec.clone();
        other_spec.set_config("krea2.euler.grid_x", "2");
        let other_key = kernels.compiler().module(source).key(&other_spec)?;
        let outcome = kernels.0.build_with(1, |_| {
            // Another caller requests both the active kernel and a new one
            // while loading is paused here. No native GPU load is performed.
            std::thread::scope(|scope| {
                scope
                    .spawn(|| {
                        for _ in 0..2 {
                            let duplicate = kernels.request(source, &spec).unwrap();
                            assert_eq!(duplicate.key, pending.key);
                            kernels.request(source, &other_spec).unwrap();
                        }
                    })
                    .join()
                    .unwrap();
            });
            assert_eq!(kernels.0.locked().unwrap().waiting.len(), 1);
            Err(Error::Message("injected load failure".into()))
        });
        assert_eq!(outcome.unwrap_err().to_string(), "injected load failure");
        {
            let state = kernels.0.locked()?;
            let waiting: Vec<_> = state.waiting.iter().map(|r| r.0.as_str()).collect();
            assert_eq!(waiting, vec![pending.key.as_str(), other_key.as_str()]);
            assert!(state.active.is_empty());
        }
        let mut attempts = Vec::new();
        let outcome = kernels.0.build_with(1, |artifact| {
            attempts.push(artifact.record.key.clone());
            Err(Error::Message("injected retry failure".into()))
        });
        assert!(outcome.is_err());
        assert_eq!(attempts, vec![pending.key, other_key]);
        assert_eq!(kernels.0.locked()?.waiting.len(), 2);
        Ok(())
    }

    #[test]
    #[ignore = "requires libloomc.so, no GPU"]
    fn compile_and_load_failures_remain_requested_for_retry() -> Result<()> {
        let kernels = Kernels::new(Compiler::resolve(None)?);
        let source = include_str!("../tests/kernels/euler.loom");
        let bad = kernels.request(source, &Specialization::new("missing_export"))?;
        let good = kernels.request(source, &euler_spec())?;
        for _ in 0..2 {
            let mut attempted = Vec::new();
            let result = kernels.0.build_with(1, |artifact| {
                attempted.push(artifact.record.key.clone());
                Err(Error::Message("injected transient load failure".into()))
            });
            let error = result.unwrap_err().to_string();
            assert!(
                !error.contains("injected transient load failure"),
                "the first compile failure must be returned: {error}"
            );
            assert_eq!(
                attempted,
                vec![good.key.clone()],
                "the later request was attempted"
            );
            let state = kernels.0.locked()?;
            let waiting: Vec<_> = state.waiting.iter().map(|r| r.0.as_str()).collect();
            assert_eq!(waiting, vec![bad.key.as_str(), good.key.as_str()]);
        }
        assert!(good.built().is_none());
        Ok(())
    }

    #[test]
    #[ignore = "requires libloomc.so, no GPU"]
    fn concurrent_build_waits_for_the_active_batch_and_retries_its_failure() -> Result<()> {
        use std::sync::mpsc;
        use std::time::Duration;
        let kernels = Kernels::new(Compiler::resolve(None)?);
        let pending =
            kernels.request(include_str!("../tests/kernels/euler.loom"), &euler_spec())?;
        let (entered_tx, entered_rx) = mpsc::channel();
        let (release_tx, release_rx) = mpsc::channel();
        let (started_tx, started_rx) = mpsc::channel();
        let (done_tx, done_rx) = mpsc::channel();
        std::thread::scope(|scope| {
            let first_cache = &kernels.0;
            let first = scope.spawn(move || {
                first_cache.build_with(1, |_| {
                    entered_tx.send(()).unwrap();
                    release_rx.recv_timeout(Duration::from_secs(10)).unwrap();
                    Err(Error::Message("first load failed".into()))
                })
            });
            entered_rx.recv_timeout(Duration::from_secs(10)).unwrap();
            assert!(
                kernels.0.locked().unwrap().waiting.is_empty(),
                "batch has drained"
            );
            let second = scope.spawn(|| {
                started_tx.send(()).unwrap();
                let result = kernels
                    .0
                    .build_with(1, |_| Err(Error::Message("retry failed".into())));
                done_tx.send(result.map_err(|e| e.to_string())).unwrap();
            });
            started_rx.recv_timeout(Duration::from_secs(10)).unwrap();
            let early = done_rx.recv_timeout(Duration::from_millis(100));
            release_tx.send(()).unwrap();
            assert!(first.join().unwrap().is_err());
            second.join().unwrap();
            assert!(
                matches!(early, Err(mpsc::RecvTimeoutError::Timeout)),
                "build returned before the active batch finished"
            );
            assert_eq!(done_rx.recv().unwrap(), Err("retry failed".into()));
        });
        assert!(pending.built().is_none());
        assert_eq!(kernels.0.locked()?.waiting.len(), 1);
        Ok(())
    }

    #[test]
    #[ignore = "requires libloomc.so, no GPU"]
    fn module_cache_eviction_preserves_live_modules() -> Result<()> {
        let compiler = Compiler::with_options(
            None,
            CompilerOptions {
                module_cache_capacity: 2,
                ..Default::default()
            },
        )?;
        let first = compiler.module(include_str!("../tests/kernels/euler.loom"));
        let first_id = first.identity().to_owned();
        for i in 0..8 {
            compiler.module(&format!("// source {i}"));
            assert!(compiler.0.modules.lock().unwrap().len() <= 2);
        }
        compiler.trim();
        assert!(compiler.0.modules.lock().unwrap().is_empty());
        assert_eq!(first.identity(), first_id);
        let mut spec = Specialization::new("krea2_euler");
        spec.set_config("krea2.euler.grid_x", "1");
        spec.set_config("krea2.euler.grid_y", "1");
        assert!(!first.compile(&spec)?.bytes().is_empty());
        Ok(())
    }
}

/// Loaded kernels, keyed by the artifact their specialization compiles to.
///
/// The compiler already caches *compilation* on disk. This caches the loaded
/// executable, which is what a dispatch loop needs: turning a cached artifact
/// into a [`crate::gpu::Kernel`] is a native load every time, and a model that dispatches
/// the same kernel per block would pay it per block. Every consumer of this
/// crate wrote this cache, so it lives here instead.
///
/// The key is the artifact's own identity — the same digest the disk cache uses
/// — rather than whatever the caller derived a request from, so two requests
/// that compile to the same artifact share one executable.
///
/// Kernels are device-scoped, so a cache serves one device and refuses another.
#[derive(Clone)]
pub struct Kernels(Arc<Cache>);

struct Cache {
    compiler: Compiler,
    state: Mutex<Loaded>,
    /// Covers draining, compiling, publishing and requeuing one complete batch.
    build: Mutex<()>,
    report: Mutex<Option<Reporter>>,
}

/// Called with each artifact a [`Kernels`] builds. See [`Kernels::reporting`].
type Reporter = Arc<dyn Fn(&Artifact) + Send + Sync>;

#[derive(Default)]
struct Loaded {
    /// Set by the first request; kernels from one device are useless on another.
    device: Option<usize>,
    ready: HashMap<String, crate::Kernel>,
    /// Requested but not yet built, in request order.
    waiting: Vec<(String, String, Specialization)>,
    /// Keep drained requests known until their results are published or requeued.
    active: HashSet<String>,
}

/// A kernel that has been asked for but not yet built.
///
/// Compiling is what a cold cache costs and requests are independent, so
/// [`Kernels::request`] records one and returns this rather than stopping to
/// build it. The outstanding set is then built together, on as many threads as
/// the compiler allows, by [`Kernels::build`] or by the first
/// [`Pending::resolve`] that needs any one of them.
pub struct Pending {
    key: String,
    cache: Arc<Cache>,
    /// This handle's own copy, so a built kernel can be borrowed for as long as
    /// the handle lives. A graph records `&'g Kernel`, and the cache's own copy
    /// is behind a lock with no lifetime to lend.
    cell: std::sync::OnceLock<crate::Kernel>,
}

impl Clone for Pending {
    fn clone(&self) -> Self {
        Self {
            key: self.key.clone(),
            cache: self.cache.clone(),
            cell: self.cell.clone(),
        }
    }
}

impl Pending {
    /// The kernel, if its batch has already been built.
    ///
    /// For callers with no stream to build one on — recording into a graph,
    /// which cannot load an executable. Everywhere else wants
    /// [`Pending::resolve`].
    pub fn built(&self) -> Option<&crate::Kernel> {
        if self.cell.get().is_none() {
            let found = self.cache.locked().ok()?.ready.get(&self.key).cloned()?;
            let _ = self.cell.set(found);
        }
        self.cell.get()
    }

    /// The kernel, building everything outstanding if this is the first call
    /// that needs it.
    ///
    /// # Safety
    /// As [`Kernels::get`], for every source passed to [`Kernels::request`].
    pub unsafe fn resolve(&self, stream: &crate::Stream) -> Result<&crate::Kernel> {
        if self.built().is_some() {
            return Ok(self.cell.get().expect("just built"));
        }
        // A batch reports the first kernel in it that would not build, which is
        // not necessarily this one, so ask again before passing that failure on
        // as though it were ours.
        // Safety: the caller vouched for every requested source.
        let outcome = unsafe { self.cache.build(stream) };
        match self.built() {
            Some(_) => Ok(self.cell.get().expect("just built")),
            None => Err(outcome
                .err()
                .unwrap_or_else(|| Error::Message("kernel was never requested".into()))),
        }
    }
}

impl Kernels {
    /// Index this cache by a caller's compact request key.
    ///
    /// Source hashing and specialization construction happen only on a key
    /// miss. Distinct caller keys that name the same artifact still share one
    /// loaded executable. Include every input that changes the source or
    /// specialization in the key.
    pub fn keyed<K>(self) -> KeyedKernels<K> {
        KeyedKernels {
            kernels: self,
            keys: Mutex::new(HashMap::new()),
        }
    }

    /// An empty cache over a compiler. Clone it to share one cache.
    pub fn new(compiler: Compiler) -> Self {
        Self(Arc::new(Cache {
            compiler,
            state: Mutex::new(Loaded::default()),
            build: Mutex::new(()),
            report: Mutex::new(None),
        }))
    }

    /// Call `report` with each artifact as it is built, once, before it loads.
    ///
    /// A cache that returned only kernels would swallow what the compiler said
    /// about them: warnings, and the backend remarks that are the only warning
    /// a kernel is spilling registers. Cache hits report nothing, having built
    /// nothing.
    /// Installing a reporter updates every clone and existing pending handle.
    /// An invocation already in progress may finish using the previous reporter.
    /// The callback must not build or resolve requests on this same cache.
    pub fn reporting(self, report: impl Fn(&Artifact) + Send + Sync + 'static) -> Self {
        let previous = self
            .0
            .report
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .replace(Arc::new(report));
        // Captured values can have arbitrary destructors; drop outside the lock.
        drop(previous);
        self
    }

    /// The compiler these were built with.
    pub fn compiler(&self) -> &Compiler {
        &self.0.compiler
    }

    /// How many kernels are loaded.
    pub fn len(&self) -> usize {
        self.0.locked().map_or(0, |state| state.ready.len())
    }

    /// Whether any kernel is loaded.
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Compile `spec` from `source` if needed and load it, or return the
    /// loaded kernel. Outstanding [`Kernels::request`]s are left alone.
    ///
    /// # Safety
    /// `source` must be trusted Loom: its compiled output is loaded as native
    /// code, as for [`crate::Stream::load_artifact`].
    pub unsafe fn get(
        &self,
        stream: &crate::Stream,
        source: &str,
        spec: &Specialization,
    ) -> Result<crate::Kernel> {
        let module = self.0.compiler.module(source);
        let key = module.key(spec)?;
        // Safety: the caller vouched for this module's source.
        unsafe { self.get_module(stream, &module, key, spec) }
    }

    unsafe fn get_module(
        &self,
        stream: &crate::Stream,
        module: &Module,
        key: String,
        spec: &Specialization,
    ) -> Result<crate::Kernel> {
        {
            let mut state = self.0.locked()?;
            state.claim(stream)?;
            if let Some(kernel) = state.ready.get(&key) {
                return Ok(kernel.clone());
            }
        }
        // Share the batch publication lock so direct requests cannot load a
        // second executable while a batch or another direct miss is in flight.
        let _build = self
            .0
            .build
            .lock()
            .map_err(|_| Error::Message("kernel build poisoned".into()))?;
        if let Some(kernel) = self.0.locked()?.ready.get(&key) {
            return Ok(kernel.clone());
        }
        let artifact = module.compile(spec)?;
        self.0.reported(&artifact);
        // Safety: the caller vouched for the source this artifact came from.
        let kernel = unsafe { stream.load_artifact(&artifact) }?;
        let mut state = self.0.locked()?;
        state.waiting.retain(|(waiting, ..)| *waiting != key);
        state.ready.insert(key, kernel.clone());
        Ok(kernel)
    }

    /// Ask for a kernel without building it. Nothing is compiled until
    /// [`Kernels::build`] or the first [`Pending::resolve`], and then every
    /// outstanding request is built at once across the compiler's workspaces —
    /// so a consumer can name its whole set first and pay for it in one batch.
    pub fn request(&self, source: &str, spec: &Specialization) -> Result<Pending> {
        let key = self.0.compiler.module(source).key(spec)?;
        let mut state = self.0.locked()?;
        let known = state.ready.contains_key(&key)
            || state.active.contains(&key)
            || state.waiting.iter().any(|(waiting, ..)| *waiting == key);
        if !known {
            state
                .waiting
                .push((key.clone(), source.to_owned(), spec.clone()));
        }
        drop(state);
        Ok(Pending {
            key,
            cache: self.0.clone(),
            cell: std::sync::OnceLock::new(),
        })
    }

    /// Build every outstanding request, in one batch.
    /// Concurrent builds wait for the active batch before checking for more work.
    ///
    /// # Safety
    /// As [`Kernels::get`], for every source passed to [`Kernels::request`].
    pub unsafe fn build(&self, stream: &crate::Stream) -> Result<()> {
        // Safety: the caller vouched for every requested source.
        unsafe { self.0.build(stream) }
    }
}

impl Cache {
    fn locked(&self) -> Result<std::sync::MutexGuard<'_, Loaded>> {
        self.state
            .lock()
            .map_err(|_| Error::Message("kernel cache poisoned".into()))
    }

    fn reported(&self, artifact: &Artifact) {
        let report = self
            .report
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .clone();
        // A reporter may replace itself or inspect the cache.
        if let Some(report) = report {
            report(artifact);
        }
    }

    /// Compile off the device, then load in order.
    ///
    /// A specialization that will not build is that one's failure and no one
    /// else's, so the batch carries on and the rest is loaded. The failure goes
    /// back on the queue still wanted: dropping it would leave the handles that
    /// asked for it waiting on nothing, with no way to say why, while returning
    /// it means the next attempt reports the same failure again.
    unsafe fn build(&self, stream: &crate::Stream) -> Result<()> {
        self.build_with(stream.device_id(), |artifact| {
            // Safety: the caller vouched for every requested source.
            unsafe { stream.load_artifact(artifact) }
        })
    }

    fn build_with(
        &self,
        device: usize,
        mut load: impl FnMut(&Artifact) -> Result<crate::Kernel>,
    ) -> Result<()> {
        let _build = self
            .build
            .lock()
            .map_err(|_| Error::Message("kernel build poisoned".into()))?;
        let waiting = {
            let mut state = self.locked()?;
            state.claim_device(device)?;
            let waiting = std::mem::take(&mut state.waiting);
            state
                .active
                .extend(waiting.iter().map(|(key, ..)| key.clone()));
            waiting
        };
        if waiting.is_empty() {
            return Ok(());
        }
        let modules: Vec<Module> = waiting
            .iter()
            .map(|(_, source, _)| self.compiler.module(source))
            .collect();
        let requests: Vec<(&Module, &Specialization)> = modules
            .iter()
            .zip(waiting.iter().map(|(_, _, spec)| spec))
            .collect();
        let built = self.compiler.compile_all(&requests);

        let mut failure = None;
        let mut unbuilt = Vec::new();
        for (request, outcome) in waiting.into_iter().zip(built) {
            let loaded = outcome.and_then(|artifact| {
                self.reported(&artifact);
                load(&artifact)
            });
            match loaded {
                Ok(kernel) => {
                    self.locked()?.ready.insert(request.0.clone(), kernel);
                }
                Err(error) => {
                    unbuilt.push(request);
                    failure = failure.or(Some(error));
                }
            }
        }
        {
            let mut state = self.locked()?;
            unbuilt.append(&mut state.waiting);
            state.waiting = unbuilt;
            state.active.clear();
        }
        match failure {
            Some(error) => Err(error),
            None => Ok(()),
        }
    }
}

/// A caller-key index over HRX's artifact-keyed loaded kernels.
///
/// Create with [`Kernels::keyed`]. A hit checks device ownership and returns a
/// shared executable without hashing source, serializing a specialization or
/// invoking the request factory. The index stores artifact identities, not a
/// second collection of executables.
pub struct KeyedKernels<K> {
    kernels: Kernels,
    keys: Mutex<HashMap<K, String>>,
}

impl<K: Eq + std::hash::Hash> KeyedKernels<K> {
    /// The underlying batch cache and its compiler.
    pub fn kernels(&self) -> &Kernels {
        &self.kernels
    }

    /// Index a pending request without compiling. Hits do not call the factory
    /// or hash source. Failed compilation remains queued for a later retry.
    /// The factory must not reenter this index.
    ///
    /// # Safety
    /// Equal keys must describe the same source and specialization. Loading is
    /// still governed by [`Pending::resolve`]'s trusted-code contract.
    pub unsafe fn request_or_insert_with<S: AsRef<str>>(
        &self,
        stream: &crate::Stream,
        key: K,
        request: impl FnOnce(&K) -> Result<(S, Specialization)>,
    ) -> Result<Pending> {
        let mut keys = self
            .keys
            .lock()
            .map_err(|_| Error::Message("kernel key index poisoned".into()))?;
        self.kernels.0.locked()?.claim(stream)?;
        if let Some(artifact) = keys.get(&key) {
            return Ok(Pending {
                key: artifact.clone(),
                cache: self.kernels.0.clone(),
                cell: std::sync::OnceLock::new(),
            });
        }
        let (source, spec) = request(&key)?;
        let pending = self.kernels.request(source.as_ref(), &spec)?;
        keys.insert(key, pending.key.clone());
        Ok(pending)
    }

    /// Return the kernel named by `key`, constructing its request only on a miss.
    ///
    /// Failed requests are not indexed and can be retried. The factory is run
    /// under the index lock; it and the reporting callback must not reenter
    /// this index. The underlying cache's device restriction applies to hits too.
    ///
    /// # Safety
    /// As [`Kernels::get`]. Equal keys must always describe the same source and
    /// specialization, including compiler-report settings. The factory is not
    /// evaluated on a hit, so changing it does not change a cached kernel.
    pub unsafe fn get_or_insert_with<'s>(
        &self,
        stream: &crate::Stream,
        key: K,
        request: impl FnOnce(&K) -> Result<(&'s str, Specialization)>,
    ) -> Result<crate::Kernel> {
        let mut keys = self
            .keys
            .lock()
            .map_err(|_| Error::Message("kernel key index poisoned".into()))?;
        {
            let mut state = self.kernels.0.locked()?;
            state.claim(stream)?;
            if let Some(artifact) = keys.get(&key) {
                if let Some(kernel) = state.ready.get(artifact) {
                    return Ok(kernel.clone());
                }
                let pending = Pending {
                    key: artifact.clone(),
                    cache: self.kernels.0.clone(),
                    cell: std::sync::OnceLock::new(),
                };
                drop(state);
                drop(keys);
                // Safety: equal keys identify the source vouched for by the caller.
                return unsafe { pending.resolve(stream) }.cloned();
            }
        }
        let (source, spec) = request(&key)?;
        let module = self.kernels.compiler().module(source);
        let artifact = module.key(&spec)?;
        // Safety: the caller vouched for the source and the request key.
        let kernel = unsafe {
            self.kernels
                .get_module(stream, &module, artifact.clone(), &spec)
        }?;
        keys.insert(key, artifact);
        Ok(kernel)
    }
}

impl Loaded {
    /// Bind this cache to a device on first use, and hold it there.
    fn claim(&mut self, stream: &crate::Stream) -> Result<()> {
        self.claim_device(stream.device_id())
    }

    fn claim_device(&mut self, device: usize) -> Result<()> {
        match self.device {
            Some(owner) if owner != device => Err(Error::Message(
                "a kernel cache serves one device: an executable loaded on another is not \
                 dispatchable here"
                    .into(),
            )),
            Some(_) => Ok(()),
            None => {
                self.device = Some(device);
                Ok(())
            }
        }
    }
}

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