nord-usb 0.6.0

//! Typed operations.
//!
//! Each primitive runs inside a [`Session`]; callers can batch by opening one
//! session and applying the primitive repeatedly. Operations include device-side
//! progress messages but omit reads used only to refresh a host UI.

use nord_format::cbin::{Cbin, RawBody};

use crate::envelope;
use crate::error::{Error, Result};
use crate::session::ReadWrite;
use crate::session::{Session, WRITE_LIMIT};
use crate::transport::Transport;
use crate::wire::{
    cmd, read_u32, ui, AllocationUnit, Bank, Dependency, Location, Message, ObjectClass, Partition,
    ProgramInfo, Service, Status,
};

/// Query the inventory for the class the session was opened with.
///
/// **Read-only.** It sends one request and reads counters back; nothing on the
/// instrument changes. That makes it the safe way to prove the whole stack works
/// against real hardware.
pub async fn status<T: Transport, C>(session: &mut Session<'_, T, C>) -> Result<Status> {
    let class = session.class();
    let resp = session
        .request(
            Service::Program,
            10,
            cmd::STATUS,
            &class.to_raw().to_be_bytes(),
        )
        .await?;
    Status::decode(class, &resp)
}

/// Query every class worth reporting, one transaction each.
///
/// Each class needs its own session because the class is fixed at `SESSION_OPEN`.
/// Two refusals are skipped rather than failing the sweep, because instruments differ
/// in which classes they answer for: a refused `SESSION_OPEN` ([`Error::ClassRefused`])
/// and a refused `STATUS`. Every other error, a refused `HELLO` included, propagates.
pub async fn inventory<T: Transport>(transport: &mut T) -> Result<Vec<Status>> {
    let mut out = Vec::new();
    for class in ObjectClass::INVENTORY {
        let mut session = match Session::open(transport, class).await {
            Ok(s) => s,
            Err(Error::ClassRefused { .. }) => continue,
            Err(e) => return Err(e),
        };
        let result = status(&mut session).await;
        session.commit().await?;
        match result {
            Ok(s) => out.push(s),
            Err(Error::DeviceStatus(_)) => {}
            Err(e) => return Err(e),
        }
    }
    Ok(out)
}

/// Ask the device about one slot: format tag, body length, name, body checksum.
///
/// **Read-only.**
pub async fn info<T: Transport, C>(
    session: &mut Session<'_, T, C>,
    at: Location,
) -> Result<ProgramInfo> {
    let mut args = Vec::new();
    at.write_to(&mut args);
    let resp = session
        .request(Service::Program, 10, cmd::INFO, &args)
        .await?;
    let info = ProgramInfo::decode(&resp)?;
    if info.location != at {
        return Err(Error::UnexpectedLocation {
            requested: at,
            reported: info.location,
        });
    }
    Ok(info)
}

/// Read one program off the instrument, returning the bytes of a `.ne5p` file.
///
/// **Read-only.** The body is wrapped in a `CBIN` header ([`envelope`]) so the result
/// is a real file, and the device's own CRC-32 is checked against it when the device
/// supplies one.
pub async fn read_program<T: Transport, C>(
    session: &mut Session<'_, T, C>,
    at: Location,
) -> Result<Vec<u8>> {
    let (meta, body) = transfer_out(session, at).await?;

    let file = envelope::wrap(&meta.format, at, meta.version, &body)?;
    if let Some(expected) = meta.crc32 {
        let actual = envelope::crc32(&body);
        if expected != actual {
            return Err(Error::Envelope(format!(
                "body checksum mismatch: device reported {expected:08x}, received {actual:08x}"
            )));
        }
    }
    Ok(file)
}

/// Read an entity's body off the instrument **without** wrapping it in a CBIN header.
///
/// For formats whose header layout is not yet known — notably CBIN **type-0**, the
/// legacy no-CRC variant — wrapping would fabricate a header rather than reproduce one.
/// This returns exactly the bytes the device sent, which is the safe thing to archive.
pub async fn read_body<T: Transport, C>(
    session: &mut Session<'_, T, C>,
    at: Location,
) -> Result<Vec<u8>> {
    Ok(transfer_out(session, at).await?.1)
}

/// Body bytes to ask for in one `READ`. A body larger than this arrives across several
/// requests with the offset advancing by exactly this much and a short final chunk.
///
/// NSM asks for `32720`. Inferred from specimens; not confirmed on hardware.
///
/// Unexplained: some objects are read at `32726` throughout — a fixed 6-byte difference
/// that is per object, not per chunk. Both fit inside one `READ_BUFFER`, and the host
/// chooses the number, so the smaller is used uniformly.
const READ_CHUNK: u32 = 32720;

/// Body bytes per `WRITE_DATA` frame. The whole frame must stay under the device's
/// max transfer; an oversized frame wedges the instrument until a power cycle.
const WRITE_CHUNK: usize = 32720;

/// Fault-injection overrides; absent variables keep captured sizes, invalid values fail.
#[cfg(any(feature = "fault-injection", test))]
fn parse_chunk(name: &str, value: Option<&str>, default: u64) -> Result<u64> {
    let Some(value) = value else {
        return Ok(default);
    };
    value
        .parse()
        .ok()
        .filter(|&size| size > 0)
        .ok_or_else(|| Error::InvalidArgument(format!("{name} must be a positive integer")))
}

#[cfg(feature = "fault-injection")]
fn chunk_override(name: &str, default: u64) -> Result<u64> {
    match std::env::var(name) {
        Ok(value) => parse_chunk(name, Some(&value), default),
        Err(std::env::VarError::NotPresent) => Ok(default),
        Err(std::env::VarError::NotUnicode(_)) => {
            Err(Error::InvalidArgument(format!("{name} must be UTF-8")))
        }
    }
}

#[cfg(feature = "fault-injection")]
fn read_chunk() -> Result<u32> {
    let size = chunk_override("NORD_READ_CHUNK", READ_CHUNK.into())?;
    u32::try_from(size).map_err(|_| Error::InvalidArgument("NORD_READ_CHUNK exceeds u32".into()))
}
#[cfg(not(feature = "fault-injection"))]
fn read_chunk() -> Result<u32> {
    Ok(READ_CHUNK)
}

#[cfg(feature = "fault-injection")]
fn write_chunk() -> Result<usize> {
    let size = chunk_override("NORD_WRITE_CHUNK", WRITE_CHUNK as u64)?;
    usize::try_from(size)
        .map_err(|_| Error::InvalidArgument("NORD_WRITE_CHUNK exceeds usize".into()))
}
#[cfg(not(feature = "fault-injection"))]
fn write_chunk() -> Result<usize> {
    Ok(WRITE_CHUNK)
}

/// Read the metadata and body through the device's chunked transfer sequence.
async fn transfer_out<T: Transport, C>(
    session: &mut Session<'_, T, C>,
    at: Location,
) -> Result<(ProgramInfo, Vec<u8>)> {
    let chunk_size = read_chunk()?;
    let meta = info(session, at).await?;

    session.notify(&ui::label("Uploading...")?).await?;

    let mut args = Vec::new();
    at.write_to(&mut args);
    session
        .request(Service::Program, 10, cmd::BEGIN_READ, &args)
        .await?;

    // Clamp allocation from the device-supplied length; large valid bodies grow by chunk.
    let mut body = Vec::with_capacity((meta.body_len as usize).min(1 << 20));
    let mut painted = None;
    while (body.len() as u32) < meta.body_len {
        let offset = body.len() as u32;
        let want = chunk_size.min(meta.body_len - offset);

        let mut req = args.clone();
        req.extend_from_slice(&offset.to_be_bytes());
        req.extend_from_slice(&want.to_be_bytes());
        let resp = session
            .request(Service::Program, 10, cmd::READ, &req)
            .await?;

        let chunk = read_payload(resp.payload(), at, offset, want)?;
        body.extend_from_slice(chunk);

        // Progress moves only at whole percentages.
        let pct = (body.len() as u64 * 100 / (meta.body_len.max(1)) as u64) as u16;
        if painted != Some(pct) {
            session.notify(&ui::percent(pct)).await?;
            painted = Some(pct);
        }
    }

    // A zero-length body never enters the loop, so the bar would otherwise never be
    // cleared off the instrument's display.
    if painted != Some(100) {
        session.notify(&ui::percent(100)).await?;
    }
    session
        .request(Service::Program, 10, cmd::END_TRANSFER, &args)
        .await?;
    Ok((meta, body))
}

fn read_payload(payload: &[u8], at: Location, offset: u32, length: u32) -> Result<&[u8]> {
    let echoed = (
        read_u32(payload, 0)?,
        read_u32(payload, 4)?,
        read_u32(payload, 8)?,
        read_u32(payload, 12)?,
    );
    let expected = (at.bank, at.slot, offset, length);
    if echoed != expected {
        return Err(Error::Transport(format!(
            "READ response echoed {echoed:?}, expected {expected:?}"
        )));
    }
    let body = &payload[16..];
    if body.len() != length as usize {
        return Err(Error::Transport(format!(
            "asked for {length} bytes at offset {offset} but the device sent {}",
            body.len()
        )));
    }
    Ok(body)
}

/// Bound on polling `0x26` for the cleaning pass, which normally finishes within a
/// second; the headroom is for a heavily churned library.
const CLEANING_POLLS: u32 = 120;
const CLEANING_POLL_SPACING: std::time::Duration = std::time::Duration::from_millis(250);

/// Reclaim `blocks` of library space and wait for the pass to finish ("Cleaning..."
/// on the display). Writing before it finishes is refused `0x1e`.
async fn clean_library<T: Transport>(
    session: &mut Session<'_, T, ReadWrite>,
    blocks: u32,
) -> Result<()> {
    session.notify(&ui::label("Cleaning...")?).await?;
    session.notify(&ui::percent(0)).await?;
    session
        .request(
            Service::Program,
            10,
            cmd::WRITE_PREPARE,
            &blocks.to_be_bytes(),
        )
        .await?;

    let mut painted = Some(0);
    for polls in 0..CLEANING_POLLS {
        if polls > 0 {
            crate::sleep::sleep(CLEANING_POLL_SPACING).await;
        }
        let resp = session
            .request(Service::Program, 10, cmd::WRITE_PREPARE_2, &[])
            .await?;
        let (requested, done, running) = cleaning_progress(resp.payload())?;
        // Ready is `running` returning to 0; `done` can end above the request, so the
        // bar is clamped.
        if running == 0 {
            if painted != Some(100) {
                session.notify(&ui::percent(100)).await?;
            }
            return Ok(());
        }
        let pct = (done as u64 * 100 / requested.max(1) as u64).min(99) as u16;
        if painted != Some(pct) {
            session.notify(&ui::percent(pct)).await?;
            painted = Some(pct);
        }
    }
    Err(Error::Transport(format!(
        "the library's cleaning pass did not report ready within {} polls",
        CLEANING_POLLS
    )))
}

/// The `[requested, done, running]` words a cleaning-progress reply carries.
fn cleaning_progress(payload: &[u8]) -> Result<(u32, u32, u32)> {
    Ok((
        read_u32(payload, 0)?,
        read_u32(payload, 4)?,
        read_u32(payload, 8)?,
    ))
}

/// Make room for `blocks` storage blocks in a library partition, in the session that is
/// about to write. Requires a [`ReadWrite`] session.
///
/// A library write is refused `0x16` unless a prepared block exists per storage block of
/// body, so this reads [`status`] and, where `blocks` exceeds what is free, reclaims
/// exactly the shortfall out of `dirty` and waits for the pass to finish. Under what is
/// already free it sends nothing but the `STATUS` request.
///
/// `blocks` is the body's length in units of the partition's [`AllocationUnit`]; a count
/// from anywhere else sizes the reclaim wrongly. [`write`] does this for its caller.
pub async fn reserve<T: Transport>(
    session: &mut Session<'_, T, ReadWrite>,
    blocks: u32,
) -> Result<()> {
    let free = status(session).await?.free;
    if blocks > free {
        clean_library(session, blocks - free).await?;
    }
    Ok(())
}

/// Write an entity into a slot. `name` is what the slot ends up called — the file
/// carries none, and a placeholder becomes the slot's name.
///
/// A library write is refused `0x16` without a prepared block per storage block of body,
/// so where `unit` counts blocks the [`reserve`] and the transfer share one transaction;
/// a byte-granular partition sends the transfer alone. `unit` is the partition's own
/// [`AllocationUnit`] — [`Geometry::allocation_unit`](crate::device::Geometry::allocation_unit)
/// is where one comes from — and it sizes the reclaim from the CBIN body the file
/// carries, which is shorter than the file by its header.
pub async fn write<T: Transport>(
    session: &mut Session<'_, T, ReadWrite>,
    unit: AllocationUnit,
    at: Location,
    file: &[u8],
    name: &str,
    timestamp: u32,
) -> Result<()> {
    if !unit.belongs_to(session.class().to_raw()) {
        return Err(Error::InvalidArgument(format!(
            "the allocation unit belongs to another partition, not {}",
            session.class().label()
        )));
    }
    let file = envelope::unwrap(file)?;
    if !unit.is_bytes() {
        reserve(session, unit.blocks_for(file.body.0.len())?).await?;
    }
    transfer_in(session, at, &file, name, timestamp).await
}

/// A [`cmd::BEGIN_WRITE`] argument block: the address, the body's length, the format
/// tag, the timestamp, the `0xffffffff` word, and the slot's name, length-prefixed.
///
/// `BEGIN_WRITE` is the only frame of a write that carries a name; it becomes the slot's.
pub fn begin_write_args(
    at: Location,
    body_len: usize,
    tag: &[u8; 4],
    timestamp: u32,
    name: &str,
) -> Result<Vec<u8>> {
    let body_len = u32::try_from(body_len)
        .map_err(|_| Error::InvalidArgument("the body is larger than the wire format".into()))?;
    let name_len = u32::try_from(name.len())
        .map_err(|_| Error::InvalidArgument("the name is larger than the wire format".into()))?;
    let mut args = Vec::new();
    at.write_to(&mut args);
    args.extend_from_slice(&body_len.to_be_bytes());
    args.extend_from_slice(tag);
    args.extend_from_slice(&timestamp.to_be_bytes());
    args.extend_from_slice(&u32::MAX.to_be_bytes());
    args.extend_from_slice(&name_len.to_be_bytes());
    args.extend_from_slice(name.as_bytes());
    Ok(args)
}

/// A [`cmd::WRITE_DATA`] argument block: the address, the chunk's offset and length,
/// then the chunk.
pub fn write_data_args(at: Location, offset: usize, chunk: &[u8]) -> Result<Vec<u8>> {
    let offset = u32::try_from(offset)
        .map_err(|_| Error::InvalidArgument("the offset is larger than the wire format".into()))?;
    let len = u32::try_from(chunk.len())
        .map_err(|_| Error::InvalidArgument("the chunk is larger than the wire format".into()))?;
    let mut args = Vec::new();
    at.write_to(&mut args);
    args.extend_from_slice(&offset.to_be_bytes());
    args.extend_from_slice(&len.to_be_bytes());
    args.extend_from_slice(chunk);
    Ok(args)
}

/// The write transfer itself, identical for every class.
async fn transfer_in<T: Transport>(
    session: &mut Session<'_, T, ReadWrite>,
    at: Location,
    file: &Cbin<RawBody>,
    name: &str,
    timestamp: u32,
) -> Result<()> {
    let body = &file.body.0;
    let chunk_size = write_chunk()?;

    session.notify(&ui::label("Downloading...")?).await?;

    let begin = begin_write_args(at, body.len(), &file.header.tag, timestamp, name)?;
    session
        .request(Service::Program, 10, cmd::BEGIN_WRITE, &begin)
        .await?;

    let mut offset = 0usize;
    let mut painted = None;
    while offset < body.len() {
        let end = offset.saturating_add(chunk_size).min(body.len());
        let data = write_data_args(at, offset, &body[offset..end])?;
        if end == body.len() {
            // Only the final chunk is acknowledged.
            session
                .request(Service::Program, 10, cmd::WRITE_DATA, &data)
                .await?;
        } else {
            let msg = Message::new(Service::Program, 10, cmd::WRITE_DATA, data);
            session.notify(&msg).await?;
        }
        offset = end;

        let pct = (offset as u64 * 100 / (body.len().max(1)) as u64) as u16;
        if painted != Some(pct) {
            session.notify(&ui::percent(pct)).await?;
            painted = Some(pct);
        }
    }

    if painted != Some(100) {
        session.notify(&ui::percent(100)).await?;
    }

    let mut args = Vec::new();
    at.write_to(&mut args);
    session
        .request(Service::Program, 10, cmd::END_TRANSFER, &args)
        .await?;
    Ok(())
}

/// Load a stored object live on the instrument ("open on device" / double-click in
/// NSM). The device switches to it immediately.
///
/// **Non-destructive** — nothing stored changes, so this needs no [`ReadWrite`] session.
/// This is the one command with inverted parity (`0x2f` request, `0x30` response).
pub async fn select<T: Transport, C>(session: &mut Session<'_, T, C>, at: Location) -> Result<()> {
    let mut args = Vec::new();
    at.write_to(&mut args);
    session
        .request(Service::Program, 10, cmd::SELECT, &args)
        .await?;
    Ok(())
}

/// Drain queued replies until the transport stays quiet.
async fn drain<T: Transport>(transport: &mut T) -> Result<()> {
    for _ in 0..RECOVER_DRAIN_CAP {
        match transport
            .read_timeout(crate::transport::READ_BUFFER, RECOVER_DRAIN_LIMIT)
            .await?
        {
            Some(_) => continue,
            None => break,
        }
    }
    Ok(())
}

/// How long to wait for a straggler before deciding the stream is quiet.
const RECOVER_DRAIN_LIMIT: std::time::Duration = std::time::Duration::from_millis(300);

/// Upper bound on stragglers, so a device that will not stop talking cannot hang this.
const RECOVER_DRAIN_CAP: usize = 16;

/// Send one frame of the recovery sequence, naming the endpoint when it is not accepted.
///
/// ⚠️ An unbounded write here blocks forever on the very instrument this exists for: a
/// stalled bulk OUT endpoint never accepts the frame and never fails either.
async fn send_recovery<T: Transport>(transport: &mut T, msg: &Message, what: &str) -> Result<()> {
    if transport.write_timeout(&msg.encode(), WRITE_LIMIT).await? {
        return Ok(());
    }
    Err(Error::Transport(format!(
        "the device did not accept {what} within {}s: bulk OUT endpoint {:#04x} is \
         stalled, and only a power cycle clears it",
        WRITE_LIMIT.as_secs(),
        crate::transport::EP_OUT
    )))
}

/// Release UI and class state left by an abandoned session.
///
/// A bare `GOODBYE` clears the UI state that makes every slot appear empty; a bare
/// `SESSION_CLOSE` clears class status `0x12`. Queued replies are drained first.
pub async fn recover<T: Transport>(transport: &mut T) -> Result<()> {
    // An unread reply leaves every subsequent request paired with its predecessor.
    drain(transport).await?;

    // ⚠️ Bounded reads: the instrument this is for is the one that has stopped
    // answering, and no reply to either frame is the expected outcome, not a failure.
    let goodbye = Message::new(Service::Ui, ui::SUBSYSTEM, ui::GOODBYE, Vec::new());
    send_recovery(transport, &goodbye, "GOODBYE").await?;
    let _ = transport
        .read_timeout(crate::transport::READ_BUFFER, RECOVER_DRAIN_LIMIT)
        .await?;

    let close = Message::new(Service::Program, 10, cmd::SESSION_CLOSE, Vec::new());
    send_recovery(transport, &close, "SESSION_CLOSE").await?;
    let _ = transport
        .read_timeout(crate::transport::READ_BUFFER, RECOVER_DRAIN_LIMIT)
        .await?;
    Ok(())
}

/// Every storage partition the device reports. **Read-only.**
///
/// The index of each entry is its object class code, so this is also the authoritative
/// answer to "what classes does this instrument have" — including the `(Native)` library
/// views that have no [`ObjectClass`] name.
pub async fn partitions<T: Transport, C>(
    session: &mut Session<'_, T, C>,
) -> Result<Vec<Partition>> {
    let resp = session
        .request(Service::Program, 10, cmd::PARTITIONS, &[])
        .await?;
    Partition::decode_all(&resp)
}

/// One partition's banks and their slot capacities. **Read-only.**
pub async fn banks<T: Transport, C>(
    session: &mut Session<'_, T, C>,
    partition: u32,
) -> Result<Vec<Bank>> {
    let resp = session
        .request(Service::Program, 10, cmd::BANKS, &partition.to_be_bytes())
        .await?;
    let payload = resp.payload();
    if payload.len() < 4 {
        return Err(Error::Truncated {
            got: payload.len(),
            need: 4,
        });
    }
    let reported = u32::from_be_bytes(payload[..4].try_into().unwrap());
    if reported != partition {
        return Err(Error::UnexpectedPartition {
            requested: partition,
            reported,
        });
    }
    Bank::decode_all(&resp)
}

/// Whether an address exists on this instrument, per the device's own geometry.
///
/// **Read-only**, and the point is that it answers *before* anything is attempted: a write
/// to a bad address otherwise fails only once the transfer is under way, and a write to an
/// occupied one is refused with status `0x4` after the caller has committed to it.
///
/// `Ok(None)` means the address is fine. `Ok(Some(reason))` explains why it is not.
///
/// [`Geometry::check_address`](crate::device::Geometry::check_address) asks the same of
/// geometry already read, and sends nothing.
pub async fn check_address<T: Transport, C>(
    session: &mut Session<'_, T, C>,
    at: Location,
) -> Result<Option<String>> {
    let banks = banks(session, session.class().to_raw()).await?;
    Ok(address_refusal(&banks, at))
}

/// Why `at` is not an address among `banks`, or `None` where it is.
///
/// The reason is in the bank names the instrument itself uses — which for pianos are
/// categories, so "no bank 7 (this class has 6: Grand, Upright, …)" is a far better
/// error than a status code.
pub(crate) fn address_refusal(banks: &[Bank], at: Location) -> Option<String> {
    let Some(bank) = banks.get(at.bank as usize) else {
        let names: Vec<&str> = banks.iter().map(|b| b.name.as_str()).collect();
        return Some(format!(
            "bank {} does not exist; this class has {} ({})",
            at.user_bank(),
            banks.len(),
            names.join(", ")
        ));
    };
    // The `(Native)` partitions report a sentinel rather than a capacity, so there is
    // nothing to check against there.
    (bank.is_bounded() && at.slot >= bank.slots).then(|| {
        format!(
            "\"{}\" holds {} slots, so slot {} is out of range",
            bank.name,
            bank.slots,
            at.user_slot()
        )
    })
}

/// The object the panel currently has loaded, for the session's class. **Read-only.**
///
/// The read half of [`select`]: together they make the player's own position addressable.
pub async fn focus<T: Transport, C>(session: &mut Session<'_, T, C>) -> Result<Location> {
    let resp = session
        .request(Service::Program, 10, cmd::FOCUS, &[])
        .await?;
    let p = resp.payload();
    if p.len() < 8 {
        return Err(Error::Truncated {
            got: p.len(),
            need: 8,
        });
    }
    Ok(Location {
        bank: u32::from_be_bytes(p[0..4].try_into().unwrap()),
        slot: u32::from_be_bytes(p[4..8].try_into().unwrap()),
    })
}

/// Device status refusing a [`cmd::NEXT_SLOT`] without the direction word. Surfaced
/// rather than swallowed: a refused walk must not pass off a partial list.
pub const ENUMERATION_DISABLED: u32 = 0x11;

/// Slot value meaning "from the bank's boundary": the bank's first occupied slot when
/// walking forward, its last when walking backward.
pub const SLOT_BOUNDARY: u32 = 0xffff_ffff;

/// Host safety budget for one occupied-slot walk. Exceeding it is an error, not a
/// truncated inventory.
pub const ENUMERATION_LIMIT: usize = 4096;

/// The next occupied slot after `at`, or `None` once the walk runs off the end.
///
/// **Read-only.** Positions inside a gap are safe to pass: the device answers with the
/// next real object rather than an error, which is what makes this an iterator over
/// content instead of over addresses. `at.slot == SLOT_BOUNDARY` starts from before
/// the bank's first slot.
pub async fn next_occupied<T: Transport, C>(
    session: &mut Session<'_, T, C>,
    at: Location,
) -> Result<Option<Location>> {
    let mut args = Vec::new();
    at.write_to(&mut args);
    // Direction, 0 = forward; omitting it is refused after any write since power-up.
    args.extend_from_slice(&0u32.to_be_bytes());
    match session
        .request(Service::Program, 10, cmd::NEXT_SLOT, &args)
        .await
    {
        Ok(resp) => {
            let p = resp.payload();
            if p.len() < 8 {
                return Err(Error::Truncated {
                    got: p.len(),
                    need: 8,
                });
            }
            Ok(Some(Location {
                bank: u32::from_be_bytes(p[0..4].try_into().unwrap()),
                slot: u32::from_be_bytes(p[4..8].try_into().unwrap()),
            }))
        }
        // Not a fault: the position asked about is past the end, which is how the walk
        // terminates. A refusal leaves the session in step, so the caller may continue.
        Err(Error::DeviceStatus(1)) => Ok(None),
        Err(e) => Err(e),
    }
}

/// Every occupied slot in the session's class, in address order.
///
/// **Read-only.** [`next_occupied`] walks *within* one bank and stops at its end, so this
/// drives it over `banks` in table order, each from [`SLOT_BOUNDARY`]. Pianos span
/// several banks and programs fill eight of them; only the sample library is flat, and
/// walking bank 0 alone silently reports a fraction of the class.
///
/// `banks` is the instrument's own answer for this class — [`banks`] on the partition
/// whose index is the class code, or [`Geometry::banks`](crate::device::Geometry::banks).
/// Nothing here guesses how many banks a class has or how far one runs: a bank ends where
/// the device ends it (status `1` to a cursor request), and its declared capacity bounds
/// how many objects it may yield.
///
/// A cursor answer that leaves the bank, repeats, goes backwards, or exceeds the declared
/// capacity is [`Error::Enumeration`]. [`ENUMERATION_LIMIT`] bounds the complete walk;
/// exhausting it is an error rather than a truncated inventory.
///
/// A refusal mid-walk — [`ENUMERATION_DISABLED`] above all — propagates as its error
/// rather than truncating the list: a partial inventory that looks complete is the one
/// result worse than none.
pub async fn occupied_slots<T: Transport, C>(
    session: &mut Session<'_, T, C>,
    banks: &[Bank],
) -> Result<Vec<Location>> {
    let mut found: Vec<Location> = Vec::new();

    for bank in banks {
        let mut at = Location {
            bank: bank.index,
            slot: SLOT_BOUNDARY,
        };
        let mut previous = None;
        let limit = match bank.is_bounded() {
            true => bank.slots,
            false => Bank::UNBOUNDED,
        };
        while let Some(next) = next_occupied(session, at).await? {
            let advanced = next.bank == bank.index
                && next.slot < limit
                && previous.is_none_or(|slot| next.slot > slot);
            if !advanced {
                return Err(Error::Enumeration {
                    bank: bank.index,
                    answered: next,
                    slots: bank.slots,
                });
            }
            if found.len() >= ENUMERATION_LIMIT {
                return Err(Error::ScanLimit {
                    bank: bank.index,
                    limit: ENUMERATION_LIMIT as u32,
                });
            }
            found.push(next);
            at = next;
            previous = Some(next.slot);
        }
    }
    Ok(found)
}

/// List the piano/sample library objects an entity depends on, as the device reports
/// them — including rows that are not dependencies at all.
///
/// **Read-only.** The returned [`Dependency`] ids match the ids the objects carry in
/// their own files, which is the bridge between wire content and file bytes.
pub async fn dependencies<T: Transport, C>(
    session: &mut Session<'_, T, C>,
    at: Location,
) -> Result<Vec<Dependency>> {
    let mut args = Vec::new();
    at.write_to(&mut args);
    let resp = session
        .request(Service::Program, 10, cmd::DEPENDENCIES, &args)
        .await?;
    let dependencies = Dependency::decode_all(&resp)?;
    let p = resp.payload();
    let reported = Location {
        bank: u32::from_be_bytes(p[0..4].try_into().unwrap()),
        slot: u32::from_be_bytes(p[4..8].try_into().unwrap()),
    };
    if reported != at {
        return Err(Error::UnexpectedLocation {
            requested: at,
            reported,
        });
    }
    Ok(dependencies)
}

/// A set list holding a reference to a program slot a caller is about to disturb.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Referrer {
    /// Where the set list itself lives.
    pub at: Location,
    /// The set list's name, as the instrument shows it.
    pub name: String,
    /// The set list's schema version *now*. ⚠️ `0` is the one worth stopping for: the
    /// device rewrites a referring set list in the current format, so a version-0 object
    /// is migrated to version 1 and there is no route back to what it was.
    pub version: u32,
    /// Which of the queried program slots this set list points at.
    pub programs: Vec<Location>,
}

/// Every set list that references one of `targets`. **Read-only.**
///
/// The session must be open on [`ObjectClass::SetList`]; the walk and every read run
/// inside it, over the `banks` [`occupied_slots`] documents.
///
/// This is what makes a program `move` describable. The instrument maintains referential
/// integrity itself: moving a program rewrites the body of **every** set list pointing at
/// it, so a move touches objects in another class that the caller never named, and one of
/// those rewrites is irreversible where [`Referrer::version`] is `0`. Nothing in the move
/// request says so, and no reply reports it afterwards — the only way to know is to ask
/// every set list first.
///
/// Cost is one `DEPENDENCIES` per occupied set list, plus one `INFO` per match. A
/// refusal mid-scan propagates rather than truncating, and so does a walk that
/// contradicts the declared geometry, for the same reason: a short list here reads as
/// "no set list is affected", which is the wrong answer to act on.
///
/// Confirmed on hardware.
pub async fn set_lists_referencing<T: Transport, C>(
    session: &mut Session<'_, T, C>,
    banks: &[Bank],
    targets: &[Location],
) -> Result<Vec<Referrer>> {
    if session.class() != ObjectClass::SetList {
        return Err(Error::InvalidArgument(
            "set-list referrers require a set-list session".into(),
        ));
    }
    let mut out = Vec::new();
    if targets.is_empty() {
        return Ok(out);
    }
    for at in occupied_slots(session, banks).await? {
        let mut programs: Vec<Location> = Vec::new();
        for l in dependencies(session, at)
            .await?
            .into_iter()
            .filter(|d| d.class == ObjectClass::Program && d.is_required())
            .filter_map(|d| d.location)
        {
            // A set list may hold the same program in more than one of its four slots.
            if targets.contains(&l) && !programs.contains(&l) {
                programs.push(l);
            }
        }
        if programs.is_empty() {
            continue;
        }
        let meta = info(session, at).await?;
        out.push(Referrer {
            at,
            name: meta.name,
            version: meta.version,
            programs,
        });
    }
    Ok(out)
}

/// Move an object from one slot to another. The device relocates it internally — no
/// body crosses the wire.
///
/// An occupied destination is **swapped, not overwritten**: its occupant ends up in the
/// source slot, byte-identical. Nothing is destroyed, and no delete-first step is needed
/// (unlike a write, which the device refuses into an occupied slot with status `0x4`).
/// Confirmed on hardware.
///
/// ⚠️ **Moving a program is not a local operation.** The device rewrites every set list
/// referencing either slot so no reference is left dangling, changing bodies the caller
/// never named — and a version-0 set list is migrated to version 1 in the process, which
/// cannot be undone by moving the program back. [`set_lists_referencing`] names them
/// before the fact; nothing in the request or the reply mentions them.
///
/// Requires a [`ReadWrite`] session. Class-generalised: works for whichever object
/// class the session opened (programs, set lists).
pub async fn move_object<T: Transport>(
    session: &mut Session<'_, T, ReadWrite>,
    from: Location,
    to: Location,
) -> Result<()> {
    let mut args = Vec::new();
    from.write_to(&mut args);
    to.write_to(&mut args);
    session
        .request(Service::Program, 10, cmd::MOVE, &args)
        .await?;
    Ok(())
}

/// Delete the object in a slot. Requires a [`ReadWrite`] session.
///
/// Sends the `"Deleting..."` progress label the instrument paints, then the delete —
/// exactly the two OUT frames NSM sends (the `O36 O26 I30` shape).
pub async fn delete<T: Transport>(
    session: &mut Session<'_, T, ReadWrite>,
    at: Location,
) -> Result<()> {
    session.notify(&ui::label("Deleting...")?).await?;
    let mut args = Vec::new();
    at.write_to(&mut args);
    session
        .request(Service::Program, 10, cmd::DELETE, &args)
        .await?;
    Ok(())
}

/// Rename the object in a slot. Requires a [`ReadWrite`] session.
///
/// The name is sent big-endian length-prefixed and unpadded — the same encoding
/// strings use everywhere on the wire.
pub async fn rename<T: Transport>(
    session: &mut Session<'_, T, ReadWrite>,
    at: Location,
    name: &str,
) -> Result<()> {
    let mut args = Vec::new();
    at.write_to(&mut args);
    let name_len = u32::try_from(name.len())
        .map_err(|_| Error::InvalidArgument("the name is larger than the wire format".into()))?;
    args.extend_from_slice(&name_len.to_be_bytes());
    args.extend_from_slice(name.as_bytes());
    session
        .request(Service::Program, 10, cmd::RENAME, &args)
        .await?;
    Ok(())
}

/// Duplicate the object at `from` into `to`. Requires a [`ReadWrite`] session.
///
/// A deep copy the device performs internally: the arguments are just the two
/// addresses, and no body crosses the wire. (NSM follows a copy with `INFO`/`DEPENDENCIES`
/// reads to repaint its browser; those are UI bookkeeping and are not sent here — see
/// the module-level note.)
pub async fn duplicate<T: Transport>(
    session: &mut Session<'_, T, ReadWrite>,
    from: Location,
    to: Location,
) -> Result<()> {
    let mut args = Vec::new();
    from.write_to(&mut args);
    to.write_to(&mut args);
    session
        .request(Service::Program, 10, cmd::COPY, &args)
        .await?;
    Ok(())
}

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

    #[test]
    fn a_read_chunk_must_echo_its_request() {
        let at = Location { bank: 2, slot: 3 };
        let mut payload = Vec::new();
        for word in [at.bank, at.slot, 40, 3] {
            payload.extend_from_slice(&word.to_be_bytes());
        }
        payload.extend_from_slice(&[1, 2, 3]);
        assert_eq!(read_payload(&payload, at, 40, 3).unwrap(), [1, 2, 3]);

        payload[3] ^= 1;
        assert!(read_payload(&payload, at, 40, 3).is_err());
    }

    #[test]
    fn invalid_chunk_overrides_are_refused() {
        assert!(parse_chunk("NORD_READ_CHUNK", Some("0"), READ_CHUNK.into()).is_err());
        assert!(parse_chunk("NORD_WRITE_CHUNK", Some("bad"), WRITE_CHUNK as u64).is_err());
        assert_eq!(
            parse_chunk("NORD_READ_CHUNK", None, READ_CHUNK.into()).unwrap(),
            READ_CHUNK.into()
        );
    }

    #[test]
    fn cleaning_progress_requires_all_three_words() {
        let err = cleaning_progress(&[0; 11]).expect_err("a partial cleaning reply");
        assert!(matches!(err, Error::Truncated { got: 11, need: 12 }));
    }

    /// A transport that never accepts a frame and never says so, which is the state a
    /// stalled bulk OUT endpoint leaves the instrument in.
    struct Stalled;

    impl Transport for Stalled {
        async fn write(&mut self, _buf: &[u8]) -> Result<()> {
            panic!("recovery frames must carry a deadline");
        }

        async fn read(&mut self, _max: usize) -> Result<Vec<u8>> {
            panic!("recovery reads must carry a deadline");
        }

        async fn write_timeout(
            &mut self,
            _buf: &[u8],
            _limit: std::time::Duration,
        ) -> Result<bool> {
            Ok(false)
        }

        async fn read_timeout(
            &mut self,
            _max: usize,
            _limit: std::time::Duration,
        ) -> Result<Option<Vec<u8>>> {
            Ok(None)
        }
    }

    #[test]
    fn recover_names_the_stalled_endpoint_instead_of_waiting_forever() {
        let err = pollster::block_on(recover(&mut Stalled)).expect_err("the write is refused");
        let message = err.to_string();
        assert!(message.contains("GOODBYE"), "{message}");
        assert!(message.contains("0x03"), "{message}");
    }
}