hrx-rs 0.10.1

//! Explicit hardware suite: cargo test --all-features --test gpu -- --ignored
use hrx::{Constants, Device, Stream};

/// Counts heap allocations while armed, so a test can assert that a hot path
/// stays on the stack instead of only intending to.
mod counting {
    use std::alloc::{GlobalAlloc, Layout, System};
    use std::cell::Cell;
    use std::sync::atomic::{AtomicUsize, Ordering};
    thread_local! { static ARMED: Cell<bool> = const { Cell::new(false) }; }
    pub static COUNT: AtomicUsize = AtomicUsize::new(0);
    pub struct Counting;
    unsafe impl GlobalAlloc for Counting {
        unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
            if ARMED.try_with(|armed| armed.get()).unwrap_or(false) {
                COUNT.fetch_add(1, Ordering::Relaxed);
            }
            unsafe { System.alloc(layout) }
        }
        unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
            unsafe { System.dealloc(ptr, layout) }
        }
    }
    /// Allocations performed by `body`.
    pub fn count(body: impl FnOnce()) -> usize {
        COUNT.store(0, Ordering::Relaxed);
        ARMED.with(|armed| armed.set(true));
        body();
        ARMED.with(|armed| armed.set(false));
        COUNT.load(Ordering::Relaxed)
    }
}
#[global_allocator]
static ALLOCATOR: counting::Counting = counting::Counting;

#[test]
#[ignore = "requires gfx1151"]
fn partial_host_writes_preserve_gpu_written_cache_line_neighbors() -> hrx::Result<()> {
    let mut stream = Stream::open()?;
    let buffer = stream.allocate(128)?;
    let mut actual = [0; 128];
    for value in 1..=32u8 {
        // Cache the old host view before the GPU replaces it.
        stream.read_blocking(buffer.binding(), &mut actual)?;
        stream.fill(buffer.binding(), value)?;
        stream.upload_blocking(buffer.slice(17, 1), &[0xfe])?;
        stream.read_blocking(buffer.binding(), &mut actual)?;
        let mut expected = [value; 128];
        expected[17] = 0xfe;
        assert_eq!(actual, expected);
    }
    Ok(())
}

#[test]
#[ignore = "requires gfx1151"]
fn nested_views_bound_stream_and_graph_operations() -> hrx::Result<()> {
    let mut stream = Stream::open()?;
    let source = stream.allocate(64)?;
    let destination = stream.allocate(64)?;
    let src = source.binding().slice(8, 32)?.slice(4, 16)?;
    let dst = destination.binding().slice(24, 24)?.slice(8, 16)?;
    assert_eq!(src.offset(), 12);
    assert_eq!(dst.offset(), 32);
    assert_eq!(src.len(), 16);
    assert!(std::ptr::eq(src.owner(), &source));
    assert_eq!(src.slice(16, 0)?.offset(), 28);
    assert!(src.slice(16, 0)?.is_empty());
    // These would fit in the allocation, but escape the parent view.
    assert!(src.slice(15, 2).is_err());
    assert!(src.slice(17, 0).is_err());
    assert!(src.slice(usize::MAX, 2).is_err());
    assert!(src.slice(1, usize::MAX).is_err());
    assert!(stream.upload(src, &[0; 17]).is_err());
    assert!(stream.upload_blocking(src, &[0; 17]).is_err());
    assert!(stream.copy(dst.slice(0, 15)?, src).is_err());
    let empty = src.slice(16, 0)?;
    assert!(stream.fill(empty, 1).is_err());
    assert!(stream.copy(empty, empty).is_err());
    stream.upload(empty, &[])?;

    stream.fill(source.binding(), 0x11)?;
    stream.fill(destination.binding(), 0x22)?;
    stream.fill(src, 0x33)?;
    stream.upload(src.slice(4, 4)?, &[1, 2, 3, 4])?;
    stream.upload_blocking(src.slice(8, 4)?, &[9, 10, 11, 12])?;
    stream.copy(dst, src)?;
    let mut expected = [0x11; 64];
    expected[12..28].fill(0x33);
    expected[16..20].copy_from_slice(&[1, 2, 3, 4]);
    expected[20..24].copy_from_slice(&[9, 10, 11, 12]);
    let mut actual = [0; 64];
    stream.read_blocking(source.binding(), &mut actual)?;
    assert_eq!(actual, expected);
    let mut copied = [0x22; 64];
    copied[32..48].copy_from_slice(&expected[12..28]);
    stream.read_blocking(destination.binding(), &mut actual)?;
    assert_eq!(actual, copied);

    let mut graph = stream.graph()?;
    assert!(graph.fill(&[], empty, 1).is_err());
    assert!(graph.copy(&[], empty, empty).is_err());
    assert!(graph.copy(&[], dst.slice(0, 15)?, src).is_err());
    // The copy reads what the fill writes, so the edge is declared, not implied.
    let filled = graph.fill(&[], src, 0x44)?;
    graph.copy(&[filled], dst, src)?;
    let mut exec = graph.finish()?;
    stream.launch(&mut exec)?;
    expected[12..28].fill(0x44);
    copied[32..48].fill(0x44);
    stream.read_blocking(source.binding(), &mut actual)?;
    assert_eq!(actual, expected);
    stream.read_blocking(destination.binding(), &mut actual)?;
    assert_eq!(actual, copied);
    Ok(())
}

#[test]
#[ignore = "requires gfx1151"]
fn independent_streams_move_between_threads() -> hrx::Result<()> {
    let device = hrx::Device::open(0)?;
    let streams = (0..32)
        .map(|_| device.stream())
        .collect::<hrx::Result<Vec<_>>>()?;
    let workers: Vec<_> = streams
        .into_iter()
        .enumerate()
        .map(|(index, mut stream)| {
            std::thread::spawn(move || -> hrx::Result<_> {
                let source = stream.allocate(4096)?;
                let destination = stream.allocate(4096)?;
                let expected = vec![index as u8 + 1; 4096];
                stream.upload(source.binding(), &expected)?;
                stream.copy(destination.binding(), source.binding())?;
                drop(source);
                let readback = stream.read(destination.binding())?;
                drop(destination);
                // Move pending commands and their retained storage back to the caller.
                Ok((stream, readback, expected))
            })
        })
        .collect();
    for worker in workers {
        let (mut stream, readback, expected) = worker.join().expect("stream worker panicked")?;
        assert_eq!(readback.wait(&mut stream)?, expected);
    }
    Ok(())
}

#[test]
#[ignore = "requires gfx1151"]
fn stream_queue_count_is_shared_by_device_handles() -> hrx::Result<()> {
    for count in [1, 2, 8] {
        let device = Device::open(0)?;
        assert!(device.set_stream_queue_count(0).is_err());
        device.set_stream_queue_count(count)?;
        assert_eq!(device.clone().stream_queue_count()?, count);
        assert_eq!(Device::open(0)?.stream_queue_count()?, count);
        let mut streams = (0..32)
            .map(|_| Stream::open())
            .collect::<hrx::Result<Vec<_>>>()?;
        device.set_stream_queue_count(count)?;
        assert!(matches!(
            device.set_stream_queue_count(count + 1),
            Err(hrx::Error::Busy(_))
        ));
        for (index, stream) in streams.iter_mut().enumerate() {
            let buffer = stream.allocate(64)?;
            stream.fill(buffer.binding(), index as u8)?;
            let mut bytes = [0; 64];
            stream.read_blocking(buffer.binding(), &mut bytes)?;
            assert_eq!(bytes, [index as u8; 64]);
        }
        drop(streams);
        assert!(matches!(
            device.set_stream_queue_count(count + 1),
            Err(hrx::Error::Busy(_))
        ));
    }
    Ok(())
}

#[test]
#[ignore = "requires gfx1151"]
fn many_streams_share_queues_and_exchange_events() -> hrx::Result<()> {
    let device = Device::open(0)?;
    // Reopening a device and Stream::open must join the same bounded pool.
    let mut streams = (0..128)
        .map(|index| match index % 3 {
            0 => device.clone().stream(),
            1 => Device::open(0)?.stream(),
            _ => Stream::open(),
        })
        .collect::<hrx::Result<Vec<_>>>()?;
    let identities: std::collections::HashSet<_> = streams.iter().map(Stream::id).collect();
    assert_eq!(identities.len(), streams.len());
    let buffer = streams[0].allocate(64)?;
    let mut ready = streams[0].record_event()?;
    // A stride of one pool's size exercises events between streams on the same
    // queue as well as between queues when advancing to the next offset.
    let queues = device.stream_queue_count()?;
    for offset in 0..queues {
        for index in (offset..streams.len()).step_by(queues) {
            let stream = &mut streams[index];
            stream.wait_event(&ready)?;
            stream.fill(buffer.binding(), index as u8)?;
            ready = stream.record_event()?;
        }
    }
    ready.synchronize()?;
    streams[0].wait_event(&ready)?;
    let mut actual = [0; 64];
    streams[0].read_blocking(buffer.binding(), &mut actual)?;
    assert_eq!(actual, [127; 64]);
    // Retained events and buffers keep their execution resources alive.
    drop(streams);
    ready.synchronize()?;
    let mut replacement = device.stream()?;
    replacement.fill(buffer.binding(), 42)?;
    replacement.read_blocking(buffer.binding(), &mut actual)?;
    assert_eq!(actual, [42; 64]);
    Ok(())
}

#[test]
#[ignore = "requires gfx1151"]
fn copy_and_compute_streams_exchange_event_ordered_buffers() -> hrx::Result<()> {
    let device = hrx::Device::open(0)?;
    let mut upload = device.stream()?;
    let mut compute = device.stream()?;
    let source = upload.allocate(4096)?;
    // All reads and overwrites below are ordered in both directions by events.
    let result = compute.allocate(4096)?;
    let mut readbacks = Vec::new();
    let mut last = None;
    for value in 1..=16 {
        upload.upload(source.binding(), &[value; 4096])?;
        let ready = upload.record_event()?;
        compute.wait_event(&ready)?;
        drop(ready); // The queued dependency owns its native semaphore.
        compute.copy(result.binding(), source.binding())?;
        let consumed = compute.record_event()?;
        upload.wait_event(&consumed)?;
        last = Some(consumed);
        readbacks.push(compute.read(result.binding())?);
    }
    last.as_ref().unwrap().synchronize()?;
    assert!(last.as_ref().unwrap().is_complete()?);
    for (i, readback) in readbacks.into_iter().enumerate() {
        assert_eq!(readback.wait(&mut compute)?, vec![(i + 1) as u8; 4096]);
    }
    drop(source);
    upload.synchronize()?;
    compute.synchronize()?;
    Ok(())
}
#[test]
#[ignore = "requires gfx1151 and prepared runtime bundle"]
fn queued_storage_views_and_replay() -> hrx::Result<()> {
    let mut stream = Stream::open()?;
    let source = stream.allocate(4096)?;
    let result = stream.allocate(4096)?;
    let host: Vec<u8> = (0..4096).map(|i| (i % 251) as u8).collect();
    stream.upload(source.binding(), &host)?;
    drop(host);
    stream.copy(result.binding(), source.binding())?;
    let mut submission = stream.submit()?;
    let _ = submission.is_complete()?;
    // Completion tokens are optional. Forgetting one cannot free staging.
    #[allow(clippy::forget_non_drop)] // Protect the contract if Submission later gains Drop.
    std::mem::forget(submission);
    drop(source);
    let mut bytes = vec![0; 4096];
    stream.read_blocking(result.binding(), &mut bytes)?;
    assert_eq!(
        bytes,
        (0..4096).map(|i| (i % 251) as u8).collect::<Vec<_>>()
    );
    let readback = stream.read(result.binding())?;
    assert_eq!(readback.wait(&mut stream)?, bytes);
    let abandoned = stream.read(result.binding())?;
    drop(abandoned);
    stream.synchronize()?;
    assert!(result.try_slice(usize::MAX, 1).is_err());
    assert!(result.binding().slice(usize::MAX, 1).is_err());
    let other = stream.allocate(4096)?;
    let mut graph = stream.graph()?;
    let filled = graph.fill(&[], result.binding(), 0x3c)?;
    graph.copy(&[filled], other.binding(), result.binding())?;
    let mut exec = graph.finish()?;
    for _ in 0..4 {
        stream.launch(&mut exec)?;
        stream.read_blocking(other.binding(), &mut bytes)?;
        assert!(bytes.iter().all(|b| *b == 0x3c));
    }
    Ok(())
}
#[test]
#[ignore = "requires gfx1151 and Loom compiler"]
fn prepared_binding_kernel_and_graph_match() -> hrx::Result<()> {
    let compiler = hrx::loom::Compiler::resolve(None)?;
    let module = compiler.module(include_str!("kernels/euler.loom"));
    let mut request = hrx::loom::Specialization::new("krea2_euler");
    request.set_config("krea2.euler.grid_x", "1");
    request.set_config("krea2.euler.grid_y", "1");
    let artifact = module.compile(&request)?;
    let path = artifact.path().to_path_buf();
    let mut stream = Stream::open()?;
    let kernel = unsafe { stream.load_artifact(&artifact)? };
    drop(artifact);
    drop(module);
    drop(compiler);
    let sample = stream.allocate(512)?;
    let velocity = stream.allocate(512)?;
    let ones: Vec<u8> = (0..256).flat_map(|_| 0x3f80u16.to_le_bytes()).collect();
    stream.upload_blocking(sample.binding(), &ones)?;
    stream.upload_blocking(velocity.binding(), &ones)?;
    let mut constants = Constants::new();
    // This source has precisely (index, f32). Loom's index legalization is
    // reflected in its aggregate byte count, not a general width inference.
    match kernel.info().constant_byte_length {
        8 => constants.push(256u32)?,
        12 => constants.push(256u64)?,
        n => panic!("unexpected Euler constants: {n}"),
    }
    constants.push(0.5f32)?;
    unsafe {
        assert!(
            stream
                .dispatch(
                    &kernel,
                    [1; 3],
                    [128, 1, 1],
                    &constants,
                    &[sample.binding(), velocity.binding()],
                )
                .unwrap_err()
                .to_string()
                .contains("compiled size")
        );
        stream.dispatch(
            &kernel,
            [1; 3],
            [256, 1, 1],
            &constants,
            &[sample.binding(), velocity.binding()],
        )?;
    }
    let mut output = vec![0; 512];
    stream.read_blocking(sample.binding(), &mut output)?;
    assert!(
        output
            .as_chunks::<2>()
            .0
            .iter()
            .all(|b| u16::from_le_bytes([b[0], b[1]]) == 0x3fc0)
    );
    let mut graph = stream.graph()?;
    unsafe {
        assert!(
            graph
                .dispatch(
                    &[],
                    &kernel,
                    [1; 3],
                    [128, 1, 1],
                    &constants,
                    &[sample.binding(), velocity.binding()],
                )
                .unwrap_err()
                .to_string()
                .contains("compiled size")
        );
        let copied_constants = constants.clone();
        graph.dispatch(
            &[],
            &kernel,
            [1; 3],
            [256, 1, 1],
            &copied_constants,
            &[sample.binding(), velocity.binding()],
        )?;
    }
    let mut exec = graph.finish()?;
    stream.launch(&mut exec)?;
    stream.read_blocking(sample.binding(), &mut output)?;
    assert!(
        output
            .as_chunks::<2>()
            .0
            .iter()
            .all(|b| u16::from_le_bytes([b[0], b[1]]) == 0x4000)
    );
    // Buffers are device-scoped: a second stream on the same device may dispatch
    // against them, ordered by an event rather than refused by an ownership check.
    let mut foreign = Stream::open()?;
    let ready = stream.record_event()?;
    foreign.wait_event(&ready)?;
    unsafe {
        foreign.dispatch(
            &kernel,
            [1; 3],
            [256, 1, 1],
            &constants,
            &[sample.binding(), velocity.binding()],
        )?;
        let mut graph = foreign.graph()?;
        graph.dispatch(
            &[],
            &kernel,
            [1; 3],
            [256, 1, 1],
            &constants,
            &[sample.binding(), velocity.binding()],
        )?;
        let mut recorded = graph.finish()?;
        foreign.launch(&mut recorded)?;
    }
    let done = foreign.record_event()?;
    stream.wait_event(&done)?;
    stream.read_blocking(sample.binding(), &mut output)?;
    // Two more half-steps of 0.5 on top of 0x4000 (2.0): 2.5, then 3.0.
    assert!(
        output
            .as_chunks::<2>()
            .0
            .iter()
            .all(|b| u16::from_le_bytes([b[0], b[1]]) == 0x4040)
    );
    // A kernel is still refused on a device that did not load it; with one GPU
    // present the device check is exercised through Buffer, not Kernel.
    assert!(Device::open(1).is_err());
    {
        let scratch = tempfile::tempdir()?;
        let input = scratch.path().join("runner-input.bin");
        let velocity = scratch.path().join("runner-velocity.bin");
        std::fs::write(&input, &ones)?;
        std::fs::write(&velocity, &ones)?;
        let locks = tempfile::tempdir()?;
        let mut child = std::process::Command::new(env!("CARGO_BIN_EXE_hrx"))
            .arg("run")
            .arg("--hsaco")
            .arg(&path)
            .args(["--kernel", "krea2_euler", "--block", "256"])
            .args([
                if kernel.info().constant_byte_length == 8 {
                    "--i32"
                } else {
                    "--i64"
                },
                "256",
                "--f32",
                "0.5",
            ])
            .arg("--inout")
            .arg(&input)
            .arg("--in")
            .arg(&velocity)
            .env("XDG_RUNTIME_DIR", locks.path())
            .stdout(std::process::Stdio::null())
            .spawn()?;
        let pid = child.id();
        assert!(child.wait()?.success());
        assert!(!locks.path().join(format!("hrx-{pid}.lock")).exists());
        assert!(
            std::fs::read(input)?
                .as_chunks::<2>()
                .0
                .iter()
                .all(|b| u16::from_le_bytes([b[0], b[1]]) == 0x3fc0)
        );
    }
    Ok(())
}
#[test]
#[ignore = "requires gfx1151"]
fn buffers_are_device_scoped_but_graphs_stay_stream_scoped() -> hrx::Result<()> {
    let mut a = Stream::open()?;
    let mut b = Stream::open()?;
    let source = a.allocate(16)?;
    let destination = b.allocate(16)?;

    // Every safe operation accepts an allocation from either stream, because the
    // allocator is asked for any queue and executables are device-scoped. Each
    // step below is ordered against the previous one by an event, so the reads
    // observe a defined value rather than a race.
    a.fill(source.binding(), 7)?;
    let filled = a.record_event()?;
    b.wait_event(&filled)?;

    let mut bytes = [0; 16];
    b.read_blocking(source.binding(), &mut bytes)?;
    assert_eq!(bytes, [7; 16], "a foreign stream reads the allocation");

    b.copy(destination.binding(), source.binding())?;
    b.upload(source.binding(), &[1; 16])?;
    b.upload_blocking(source.binding(), &[2; 16])?;
    b.fill(source.binding(), 3)?;
    let readback = b.read(source.binding())?;
    assert_eq!(readback.wait(&mut b)?, vec![3; 16]);
    let mut graph = b.graph()?;
    let filled = graph.fill(&[], source.binding(), 4)?;
    graph.copy(&[filled], destination.binding(), source.binding())?;
    drop(graph);

    // A recorded graph is not device-scoped: it names the stream it will replay
    // on, and native replay elsewhere would reorder against that queue.
    let mut graph = a.graph()?;
    graph.fill(&[], source.binding(), 9)?;
    let mut exec = graph.finish()?;
    assert!(
        b.launch(&mut exec)
            .unwrap_err()
            .to_string()
            .contains("another stream")
    );
    let done = b.record_event()?;
    a.wait_event(&done)?;
    a.launch(&mut exec)?;
    a.read_blocking(source.binding(), &mut bytes)?;
    assert_eq!(bytes, [9; 16]);
    Ok(())
}

#[test]
#[ignore = "requires gfx1151"]
fn instantiated_graphs_own_their_resources() -> hrx::Result<()> {
    let mut stream = Stream::open()?;
    let buffer = stream.allocate(1024)?;
    let output = stream.allocate(1024)?;
    let mut graph = stream.graph()?;
    let filled = graph.fill(&[], buffer.binding(), 42)?;
    graph.copy(&[filled], output.binding(), buffer.binding())?;
    let mut exec = graph.finish()?;
    drop(buffer);
    stream.launch(&mut exec)?;
    let mut bytes = [0; 1024];
    stream.read_blocking(output.binding(), &mut bytes)?;
    assert_eq!(bytes, [42; 1024]);
    // Even an empty executable keeps the stream/device alive after escape.
    drop(output);
    drop(stream);
    drop(exec);
    Ok(())
}

/// Dropping a handle is safe with work pending everywhere else, so it must be
/// safe here. It was not: releasing an executable graph mid-replay freed native
/// structures the device was still reading, and the next unrelated submission
/// died with an AMDGPU memory access fault rather than an error.
#[test]
#[ignore = "requires gfx1151"]
fn a_graph_released_mid_replay_does_not_fault_later_work() -> hrx::Result<()> {
    let mut stream = Stream::open()?;
    let buffer = stream.allocate(64 << 20)?;
    let mut graph = stream.graph()?;
    // A chain long enough that the replay is still running when the handle goes.
    let mut last = graph.fill(&[], buffer.binding(), 1)?;
    for value in 2..64u8 {
        last = graph.fill(&[last], buffer.binding(), value)?;
    }
    let mut exec = graph.finish()?;
    stream.launch(&mut exec)?;
    // Both go while the replay runs, as a session dropping its graph beside the
    // buffers that graph recorded.
    drop(exec);
    drop(buffer);

    // Unrelated work on the same stream must still complete.
    for _ in 0..8 {
        let other = stream.allocate(4 << 20)?;
        stream.fill(other.binding(), 7)?;
        let mut bytes = vec![0u8; 4 << 20];
        stream.read_blocking(other.binding(), &mut bytes)?;
        assert!(bytes.iter().all(|b| *b == 7));
    }
    Ok(())
}

/// Distinct streams are distinguishable, and a clone of a handle is not a
/// distinct stream. Pools that reuse allocations need this: reuse is ordered by
/// the queue, so a block may only go back to the stream it came from.
/// The cache every consumer of this crate wrote for itself: a loaded kernel per
/// artifact, built eagerly or in one deferred batch.
#[test]
#[ignore = "requires gfx1151 and Loom compiler"]
fn a_kernel_cache_loads_once_and_builds_a_batch_together() -> hrx::Result<()> {
    let source = include_str!("kernels/euler.loom");
    let spec = |grid: &str| {
        let mut request = hrx::loom::Specialization::new("krea2_euler");
        request.set_config("krea2.euler.grid_x", grid);
        request.set_config("krea2.euler.grid_y", "1");
        request
    };
    let mut stream = Stream::open()?;
    let kernels = hrx::loom::Kernels::new(hrx::loom::Compiler::for_stream(None, &stream)?);
    assert!(kernels.is_empty());

    // Safety: a checked-in Loom source compiled by this crate's own compiler.
    let first = unsafe { kernels.get(&stream, source, &spec("1")) }?;
    let again = unsafe { kernels.get(&stream, source, &spec("1")) }?;
    assert_eq!(first.symbol(), again.symbol());
    assert_eq!(kernels.len(), 1, "the same specialization loads once");

    // Deferred: named now, compiled together on build.
    let two = kernels.request(source, &spec("2"))?;
    let four = kernels.request(source, &spec("4"))?;
    assert_eq!(kernels.len(), 1, "requesting builds nothing");
    assert!(two.built().is_none(), "and yields nothing until built");
    // Safety: as above.
    unsafe { kernels.build(&stream) }?;
    assert_eq!(two.built().expect("built").symbol(), "krea2_euler");
    assert_eq!(four.built().expect("built").symbol(), "krea2_euler");
    assert_eq!(kernels.len(), 3);

    // A handle builds its own batch when nobody else has: no explicit build,
    // and the one request outstanding is enough to satisfy it.
    let eight = kernels.request(source, &spec("8"))?;
    assert!(eight.built().is_none());
    // Safety: as above.
    let resolved = unsafe { eight.resolve(&stream) }?;
    assert_eq!(resolved.symbol(), "krea2_euler");
    assert_eq!(kernels.len(), 4);

    // A specialization that cannot build fails without stranding the batch, and
    // stays outstanding so the next attempt says the same thing again.
    let bad = kernels.request(source, &spec("0"))?;
    let good = kernels.request(source, &spec("16"))?;
    // Safety: as above.
    assert!(
        unsafe { kernels.build(&stream) }.is_err(),
        "grid zero is invalid"
    );
    assert!(
        good.built().is_some(),
        "a bad neighbour does not strand a good one"
    );
    assert!(bad.built().is_none());
    // Safety: as above.
    assert!(
        unsafe { kernels.build(&stream) }.is_err(),
        "and it is still wanted"
    );

    // A second cache over the same compiler still gets its own kernels, and the
    // compiler itself is the same one, not a second resolve of the library.
    let shared = hrx::loom::Compiler::for_stream(None, &stream)?;
    assert_eq!(shared.identity(), kernels.compiler().identity());

    stream.synchronize()?;
    Ok(())
}

#[test]
#[ignore = "requires gfx1151"]
fn keyed_kernel_hits_skip_the_factory_and_allocate_nothing() -> hrx::Result<()> {
    let stream = Stream::open()?;
    let source = include_str!("kernels/euler.loom");
    let spec = || {
        let mut spec = hrx::loom::Specialization::new("krea2_euler");
        spec.set_config("krea2.euler.grid_x", "1");
        spec.set_config("krea2.euler.grid_y", "1");
        spec
    };
    let reports = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let count = reports.clone();
    let kernels = hrx::loom::Kernels::new(hrx::loom::Compiler::for_stream(None, &stream)?)
        .reporting(move |_| {
            count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        });
    let keyed = kernels.clone().keyed();
    // Failed factories must leave the key available for a retry.
    let error = unsafe {
        keyed.get_or_insert_with(&stream, 1u32, |_| {
            Err(hrx::Error::Message("temporary request failure".into()))
        })
    };
    assert!(error.is_err());
    // Safety: trusted test source. Each key consistently selects this spec.
    unsafe { keyed.get_or_insert_with(&stream, 1, |_| Ok((source, spec()))) }?;
    let allocations = counting::count(|| {
        for _ in 0..100 {
            let kernel = unsafe {
                keyed.get_or_insert_with(&stream, 1, |_| panic!("factory called on a hit"))
            }
            .unwrap();
            assert_eq!(kernel.symbol(), "krea2_euler");
        }
    });
    assert_eq!(allocations, 0, "a warm key lookup must not allocate");
    // Two caller keys for one artifact still load and report only once.
    unsafe { keyed.get_or_insert_with(&stream, 2, |_| Ok((source, spec()))) }?;
    assert_eq!(kernels.len(), 1);
    assert_eq!(reports.load(std::sync::atomic::Ordering::Relaxed), 1);
    // A compilation failure also leaves no alias that could turn into a hit.
    for _ in 0..2 {
        assert!(
            unsafe {
                keyed.get_or_insert_with(&stream, 3, |_| {
                    Ok((source, hrx::loom::Specialization::new("missing_export")))
                })
            }
            .is_err()
        );
    }
    assert_eq!(kernels.len(), 1);
    Ok(())
}

#[test]
#[ignore = "requires gfx1151"]
fn streams_are_identifiable_for_per_stream_state() -> hrx::Result<()> {
    let device = hrx::Device::open(0)?;
    let first = device.stream()?;
    let second = device.stream()?;
    assert_ne!(first.id(), second.id());
    assert_eq!(first.id(), first.id());
    Ok(())
}

#[test]
#[ignore = "requires gfx1151"]
fn graphs_fork_join_and_reject_foreign_nodes() -> hrx::Result<()> {
    let mut stream = Stream::open()?;
    let left = stream.allocate(1024)?;
    let right = stream.allocate(1024)?;
    let merged = stream.allocate(2048)?;
    let mut graph = stream.graph()?;

    // Two independent fills; neither names the other, so the runtime may overlap
    // them. The copies that read them are ordered against the right producer only.
    let fill_left = graph.fill(&[], left.binding(), 0xa1)?;
    let fill_right = graph.fill(&[], right.binding(), 0xb2)?;
    assert_ne!(
        fill_left, fill_right,
        "distinct nodes have distinct handles"
    );

    // Naming a node twice is collapsed, not rejected.
    let ready = graph.join(&[fill_left, fill_right, fill_left])?;
    graph.copy(&[ready], merged.binding().slice(0, 1024)?, left.binding())?;
    graph.copy(
        &[ready],
        merged.binding().slice(1024, 1024)?,
        right.binding(),
    )?;

    // A node from another recording names nothing here.
    let mut other = stream.graph()?;
    let foreign = other.fill(&[], left.binding(), 0)?;
    assert!(
        graph
            .fill(&[foreign], right.binding(), 0)
            .unwrap_err()
            .to_string()
            .contains("another graph")
    );
    drop(other);

    let mut exec = graph.finish()?;
    stream.launch(&mut exec)?;
    let mut bytes = vec![0; 2048];
    stream.read_blocking(merged.binding(), &mut bytes)?;
    assert!(bytes[..1024].iter().all(|b| *b == 0xa1), "left branch ran");
    assert!(bytes[1024..].iter().all(|b| *b == 0xb2), "right branch ran");
    Ok(())
}

#[test]
#[ignore = "requires gfx1151"]
fn small_fan_in_recording_does_not_allocate() -> hrx::Result<()> {
    let mut stream = Stream::open()?;
    let a = stream.allocate(1024)?;
    let b = stream.allocate(1024)?;
    let mut graph = stream.graph()?;
    let first = graph.fill(&[], a.binding(), 1)?;
    let second = graph.fill(&[], b.binding(), 2)?;
    // Warm the node vector so its growth is not attributed to dependency resolution.
    for _ in 0..8 {
        graph.fill(&[first], a.binding(), 3)?;
    }
    // Recording now prepares native commands immediately. Warm that command
    // too, so the allocation check still isolates dependency resolution.
    graph.fill(&[first], b.binding(), 4)?;
    // A 16-entry dependency list is the documented stack capacity; the duplicate
    // forces the dedupe path, which is where the allocation used to live.
    let deps = [first, second, first];
    let mut recorded = Ok(first);
    let allocations = counting::count(|| {
        recorded = graph.fill(&deps, b.binding(), 4);
    });
    recorded?;
    assert_eq!(allocations, 0, "small fan-in must stay on the stack");
    let mut exec = graph.finish()?;
    stream.launch(&mut exec)?;
    let mut bytes = [0; 1024];
    stream.read_blocking(b.binding(), &mut bytes)?;
    assert_eq!(bytes, [4; 1024]);
    Ok(())
}
#[test]
#[ignore = "requires a provisioned GPU runtime"]
fn native_budget_retains_queued_and_recorded_owners_without_double_charging_adoption() {
    use hrx::residency::ResidencyManager;
    let manager = ResidencyManager::new(8192).unwrap();
    let budget = manager.budget();
    let mut stream = hrx::Stream::open()
        .unwrap()
        .with_memory_budget(budget.clone());
    let mut consumer = hrx::Stream::open().unwrap();
    let buffer = stream.allocate(4096).unwrap();
    assert!(stream.allocate(4097).is_err());
    consumer.fill(buffer.binding(), 3).unwrap();
    drop(buffer);
    assert_eq!(budget.reserved_bytes(), 4096);
    consumer.synchronize().unwrap();
    assert_eq!(budget.reserved_bytes(), 0);

    let buffer = stream.allocate(4096).unwrap();
    let mut graph = stream.graph().unwrap();
    graph.fill(&[], buffer.binding(), 7).unwrap();
    let mut graph = graph.finish().unwrap();
    drop(buffer);
    assert_eq!(budget.reserved_bytes(), 4096);
    stream.launch(&mut graph).unwrap();
    stream.synchronize().unwrap();
    assert_eq!(budget.reserved_bytes(), 4096);
    drop(graph);
    assert_eq!(budget.reserved_bytes(), 0);

    let context = hrx::inference::ModelContext::new(hrx::execution::RuntimeOptions {
        memory_budget: Some(budget.clone()),
        ..Default::default()
    })
    .unwrap();
    let buffer = stream.allocate_zeroed(4096).unwrap();
    stream.synchronize().unwrap();
    // SAFETY: initialization drained; the buffer has no graph or external alias.
    let adopted = unsafe {
        context
            .runtime()
            .adopt_gpu_buffer(buffer, hrx::execution::MemoryPlacement::GpuLocal)
    }
    .unwrap();
    assert_eq!(budget.reserved_bytes(), 4096);
    let alias = adopted.view();
    drop(adopted);
    assert_eq!(budget.reserved_bytes(), 4096);
    drop(alias);
    assert_eq!(budget.reserved_bytes(), 0);

    let tiny = ResidencyManager::new(16).unwrap();
    let mut limited = hrx::Stream::open()
        .unwrap()
        .with_memory_budget(tiny.budget());
    let buffer = limited.allocate(16).unwrap();
    assert!(limited.upload(buffer.binding(), &[1; 16]).is_err());
    drop(buffer);
    assert_eq!(tiny.statistics().reserved_bytes, 0);
}

#[test]
#[ignore = "requires gfx1151; accepts the published bridge without profiling exports"]
fn optional_profiling_preserves_ordinary_graph_replay() -> hrx::Result<()> {
    let mut stream = Stream::open()?;
    let buffer = stream.allocate(4096)?;
    let mut diagnostic = stream.graph()?;
    diagnostic.fill(&[], buffer.binding(), 0x3c)?;
    match diagnostic.finish_profiled(&["fill".into()]) {
        Ok(mut graph) => {
            let profile = stream.launch_profiled(&mut graph)?;
            assert_eq!(profile.intervals.len(), 1);
        }
        Err(hrx::Error::Unsupported(reason))
            if reason == "libamdf ABI is missing GPU profile markers"
                || reason == "libamdf ABI is missing GPU profile clock" =>
        {
            eprintln!("older bridge correctly rejected optional profiling: {reason}");
        }
        Err(error) => return Err(error),
    }
    let mut ordinary = stream.graph()?;
    ordinary.fill(&[], buffer.binding(), 0x72)?;
    let mut graph = ordinary.finish()?;
    for _ in 0..3 {
        stream.fill(buffer.binding(), 0xcd)?;
        stream.launch(&mut graph)?;
        let mut actual = [0; 4096];
        stream.read_blocking(buffer.binding(), &mut actual)?;
        assert!(actual.iter().all(|&v| v == 0x72));
    }
    Ok(())
}

#[test]
#[ignore = "requires gfx1151 and the optional native profile bridge"]
fn profiled_graph_replay_reports_device_ticks_and_preserves_results() -> hrx::Result<()> {
    let mut stream = Stream::open()?;
    let src = stream.allocate(4096)?;
    let dst = stream.allocate(4096)?;
    let mut graph = stream.graph()?;
    let fill = graph.fill(&[], src.binding(), 0x3c)?;
    graph.copy(&[fill], dst.binding(), src.binding())?;
    let mut graph = graph.finish_profiled(&["fill".into(), "copy".into()])?;
    let mut previous_end = 0;
    for execution in 1..=4 {
        stream.fill(dst.binding(), 0xcd)?;
        let profile = stream.launch_profiled(&mut graph)?;
        assert_eq!(profile.execution, execution);
        assert!(profile.frequency_hz > 0);
        assert_eq!(profile.intervals.len(), 2);
        assert!(profile.intervals[0].start_tick > previous_end);
        assert!(profile.intervals[0].end_tick <= profile.intervals[1].start_tick);
        assert!(profile.interval_union_ms > 0.);
        assert!(profile.span_ms >= profile.interval_union_ms);
        previous_end = profile.intervals[1].end_tick;
        let mut actual = [0; 4096];
        stream.read_blocking(dst.binding(), &mut actual)?;
        assert!(actual.iter().all(|&v| v == 0x3c));
        eprintln!("{}", serde_json::to_string(&profile).unwrap());
    }
    let mut invalid = stream.graph()?;
    invalid.fill(&[], src.binding(), 0)?;
    assert!(invalid.finish_profiled(&[]).is_err());
    let mut ordinary = stream.graph()?.finish()?;
    assert!(stream.launch_profiled(&mut ordinary).is_err());
    Ok(())
}

#[test]
#[ignore = "requires gfx1151 and provisioned native libraries"]
fn large_graph_replays_across_native_batch_boundaries() -> hrx::Result<()> {
    let mut stream = Stream::open()?;
    let source = stream.allocate(16)?;
    let scratch = stream.allocate(16)?;
    // 30,000 payloads exceed one native IB for ordinary and profiled graphs.
    const ROWS: usize = 10_000;
    let output = stream.allocate(ROWS * 16)?;
    for profiled in [false, true] {
        let mut graph = stream.graph()?;
        let mut after = Vec::new();
        for i in 0..ROWS {
            let fill = graph.fill(&after, source.binding(), (i % 251 + 1) as u8)?;
            let copied = graph.copy(&[fill], scratch.binding(), source.binding())?;
            let saved = graph.copy(
                &[copied],
                output.binding().slice(i * 16, 16)?,
                scratch.binding(),
            )?;
            after = vec![saved];
        }
        let mut graph = if profiled {
            graph.finish_profiled(&(0..ROWS * 3).map(|i| i.to_string()).collect::<Vec<_>>())?
        } else {
            graph.finish()?
        };
        for _ in 0..2 {
            stream.fill(output.binding(), 0)?;
            if profiled {
                let result = stream.launch_profiled(&mut graph)?;
                assert_eq!(result.intervals.len(), ROWS * 3);
                for (i, interval) in result.intervals.iter().enumerate() {
                    assert_eq!(interval.label, i.to_string());
                    assert!(interval.end_tick >= interval.start_tick);
                    if i > 0 {
                        assert!(interval.start_tick >= result.intervals[i - 1].end_tick);
                    }
                }
            } else {
                stream.launch(&mut graph)?;
            }
            let mut bytes = vec![0; ROWS * 16];
            stream.read_blocking(output.binding(), &mut bytes)?;
            for (i, row) in bytes.as_chunks::<16>().0.iter().enumerate() {
                assert_eq!(row, &[(i % 251 + 1) as u8; 16]);
            }
        }
    }
    Ok(())
}

#[test]
#[ignore = "requires Strix Halo and the native runtime bundle"]
fn graph_argument_arena_crosses_pages_and_retains_bindings() -> hrx::Result<()> {
    let mut stream = Stream::open()?;
    let compiler = hrx::loom::Compiler::resolve(None)?;
    let module = compiler.module(include_str!("kernels/euler.loom"));
    let mut request = hrx::loom::Specialization::new("krea2_euler");
    request.set_config("krea2.euler.grid_x", "1");
    request.set_config("krea2.euler.grid_y", "1");
    let artifact = module.compile(&request)?;
    let kernel = unsafe { stream.load_artifact(&artifact)? };
    const COUNT: usize = 2049;
    let output = stream.allocate(COUNT * 512)?;
    let velocity = stream.allocate(512)?;
    stream.upload_blocking(
        velocity.binding(),
        &(0..256)
            .flat_map(|_| 0x3f80u16.to_le_bytes())
            .collect::<Vec<_>>(),
    )?;
    let mut graph = stream.owned_graph()?;
    let mut after = Vec::new();
    for i in 0..COUNT {
        let mut constants = Constants::new();
        match kernel.info().constant_byte_length {
            8 => {
                constants.push(256u32)?;
            }
            12 => {
                constants.push(256u64)?;
            }
            n => panic!("Euler ABI: {n}"),
        }
        constants.push((i % 4 + 1) as f32 * 0.5)?;
        let node = unsafe {
            graph.dispatch(
                &after,
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[output.binding().slice(i * 512, 512)?, velocity.binding()],
            )?
        };
        after = vec![node];
    }
    drop(velocity);
    drop(kernel);
    drop(artifact);
    drop(module);
    drop(compiler);
    let mut exec = graph.finish()?;
    for _ in 0..2 {
        stream.fill(output.binding(), 0)?;
        stream.launch(&mut exec)?;
        let mut actual = vec![0u8; COUNT * 512];
        stream.read_blocking(output.binding(), &mut actual)?;
        for (i, row) in actual.as_chunks::<512>().0.iter().enumerate() {
            let expected = [0x3f00u16, 0x3f80, 0x3fc0, 0x4000][i % 4].to_le_bytes();
            assert!(
                row.as_chunks::<2>().0.iter().all(|v| *v == expected),
                "argument slot {i}"
            );
        }
    }
    Ok(())
}

#[test]
#[ignore = "requires a provisioned GPU runtime"]
fn owned_graph_records_without_execution_and_retains_dropped_buffers() -> hrx::Result<()> {
    let manager = hrx::residency::ResidencyManager::new(16384)?;
    let budget = manager.budget();
    let mut stream = Stream::open()?.with_memory_budget(budget.clone());
    let output = stream.allocate(4096)?;
    stream.fill(output.binding(), 0x33)?;
    let mut graph = stream.owned_graph()?;
    {
        let temporary = stream.allocate(4096)?;
        let fill = graph.fill(&[], temporary.binding(), 0x71)?;
        graph.copy(&[fill], output.binding(), temporary.binding())?;
        assert_eq!(graph.binding_bytes(), 8192);
    }
    assert_eq!(budget.reserved_bytes(), 8192);
    let mut bytes = [0; 4096];
    stream.read_blocking(output.binding(), &mut bytes)?;
    assert_eq!(bytes, [0x33; 4096], "recording must not execute commands");
    let mut graph = graph.finish()?;
    for _ in 0..2 {
        stream.fill(output.binding(), 0)?;
        stream.launch(&mut graph)?;
        stream.read_blocking(output.binding(), &mut bytes)?;
        assert_eq!(bytes, [0x71; 4096]);
    }
    stream.launch(&mut graph)?;
    drop(graph); // Wait for its final replay and release temporary backing/charge.
    assert_eq!(budget.reserved_bytes(), 4096);
    stream.read_blocking(output.binding(), &mut bytes)?;
    assert_eq!(bytes, [0x71; 4096]);
    drop(output);
    assert_eq!(budget.reserved_bytes(), 0);
    Ok(())
}

#[test]
#[ignore = "requires a provisioned GPU runtime"]
fn owned_recording_survives_stream_drop_and_rejects_foreign_replay() -> hrx::Result<()> {
    let mut graph = {
        let stream = Stream::open()?;
        let buffer = stream.allocate(16)?;
        let mut recording = stream.owned_graph()?;
        recording.fill(&[], buffer.binding(), 1)?;
        recording
    }
    .finish()?;
    assert!(Stream::open()?.launch(&mut graph).is_err());
    Ok(())
}