vj-codegen 0.2.0

//! Streaming operators (`{<<n{...}}`, `{>>{...}}`, LRM 1800-2023 §11.4.14)
//! and bit-stream casts (§6.24.3).
//!
//! A stream is the concatenation of its operands' bit-streams, most
//! significant first. `{>>{...}}` keeps that order; `{<<n{...}}` cuts it
//! into `n`-bit slices from the least significant end and emits them in
//! reverse (a short final slice ends up least significant). An operand's
//! bit-stream is its bits, except that an unpacked array streams its
//! elements left bound first -- where vj stores the lowest index least
//! significant (a module memory's rows, an array field of an unpacked
//! struct, `vj_sim::state::storage`), [`restream`] reorders.
//!
//! Assigned to a wider target, a stream is left-aligned and zero-filled on
//! the right; a stream on the left of an assignment unpacks the (left-
//! aligned) source into its operands, the inverse reordering.
//!
//! A stream with a queue, dynamic array or string operand has a runtime
//! length: its bits gather in the runtime's stream buffer
//! (`vj_runtime::collections_rt::CollOp::StreamClear` and on), which
//! also slices it and delivers it to a fixed-width, collection or string
//! target. Unpacked, such an operand takes every bit the others leave
//! (LRM 1800-2023 §11.4.14.4).

use inkwell::values::IntValue;
use sv_lang::Type;
use sv_lang::kinds::ExpressionKind;

use vj_runtime::collections_rt::CollOp;

use crate::collections::{Args, Coll, raw_call};
use crate::error::LoweringError;
use crate::expr::{LowerCtx, LoweredValue, entry_alloca, lower_expr, lower_expr_h};
use crate::ops::OpBuilder;
use crate::types::TypeWidth;
use crate::value::PackedValue;

fn op<'a, 'ctx>(cx: &'a LowerCtx<'ctx, '_, '_>) -> OpBuilder<'a, 'ctx> {
    OpBuilder {
        ctx: cx.ctx,
        b: &cx.b,
    }
}

/// `w` bits of `v` starting at bit `lo`.
pub(crate) fn bits<'ctx>(
    cx: &LowerCtx<'ctx, '_, '_>,
    v: PackedValue<'ctx>,
    lo: u32,
    w: u32,
) -> PackedValue<'ctx> {
    let op = op(cx);
    op.part_select4(
        v,
        op.i_n(v.width).const_int(u64::from(lo), false),
        cx.ctx.bool_type().const_zero(),
        w,
    )
}

/// `{<<n{v}}` on an already-concatenated stream `v`: `n`-bit slices from
/// the least significant end, the first one most significant.
fn reverse_slices<'ctx>(
    cx: &LowerCtx<'ctx, '_, '_>,
    v: PackedValue<'ctx>,
    n: u32,
) -> PackedValue<'ctx> {
    let mut parts = Vec::new();
    let mut lo = 0;
    while lo < v.width {
        let w = n.min(v.width - lo);
        parts.push(bits(cx, v, lo, w));
        lo += w;
    }
    op(cx).concat4(&parts)
}

/// The inverse of [`reverse_slices`]: `n`-bit slices from the most
/// significant end, the first one least significant.
fn unreverse_slices<'ctx>(
    cx: &LowerCtx<'ctx, '_, '_>,
    v: PackedValue<'ctx>,
    n: u32,
) -> PackedValue<'ctx> {
    let mut parts = Vec::new();
    let mut hi = v.width;
    while hi > 0 {
        let w = n.min(hi);
        parts.push(bits(cx, v, hi - w, w));
        hi -= w;
    }
    parts.reverse();
    op(cx).concat4(&parts)
}

/// The bit-stream of the stored bits `v` of a value of type `ty`
/// ([`unstream`] is the inverse): only a fixed-size unpacked array (also
/// nested in an unpacked struct) differs -- its lowest index is stored least
/// significant but streams left bound first.
pub(crate) fn restream<'ctx>(
    cx: &LowerCtx<'ctx, '_, '_>,
    v: PackedValue<'ctx>,
    ty: &Type<'_>,
) -> Result<PackedValue<'ctx>, LoweringError> {
    use vj_sim::state::storage;
    let bad = || LoweringError::unsupported_at("a streamed value's shape", &cx.fn_sym);
    if ty.is_unpacked_array() && !ty.is_dynamically_sized_array() {
        let range = ty.fixed_unpacked_array_range();
        let elem = ty.element_type().ok_or_else(bad)?;
        let ew = storage::storage_width(&elem).ok_or_else(bad)? as u32;
        let n = range.width();
        let mut parts = Vec::with_capacity(n as usize);
        // Stream order: left bound first (most significant).
        for i in 0..n {
            let pos = if range.left <= range.right {
                i
            } else {
                n - 1 - i
            };
            parts.push(restream(cx, bits(cx, v, pos * ew, ew), &elem)?);
        }
        return Ok(op(cx).concat4(&parts));
    }
    if storage::is_unpacked_aggregate(ty) {
        let fields: Vec<_> = ty.fields().collect();
        if ty.is_unpacked_union() {
            return Ok(v);
        }
        let mut parts = Vec::with_capacity(fields.len());
        for f in &fields {
            let (lo, w) = storage::field_slice(ty, f.field_index()).ok_or_else(bad)?;
            let fty = f.value_type().ok_or_else(bad)?;
            parts.push(restream(cx, bits(cx, v, lo as u32, w as u32), &fty)?);
        }
        return Ok(op(cx).concat4(&parts));
    }
    Ok(v)
}

/// A stream operand's bit-stream.
fn operand_stream<'ctx, 'd>(
    cx: &LowerCtx<'ctx, 'd, '_>,
    e: sv_lang::Expression<'d>,
) -> Result<PackedValue<'ctx>, LoweringError> {
    let ty = e
        .expr_type()
        .ok_or_else(|| LoweringError::unsupported_at("a stream operand has no type", &cx.fn_sym))?;
    if ty.is_unpacked_array()
        && let Some((elems, elem_ty)) = crate::arrays::stored_elements_in_stream_order(cx, e)?
    {
        let parts = elems
            .into_iter()
            .map(|v| restream(cx, v, &elem_ty))
            .collect::<Result<Vec<_>, _>>()?;
        return Ok(op(cx).concat4(&parts));
    }
    if ty.is_string() || ty.is_dynamically_sized_array() {
        return Err(LoweringError::unsupported_at(
            "a string or dynamically sized stream operand",
            &cx.fn_sym,
        ));
    }
    let width = ty.value_width();
    let stored = crate::real::to_storage(cx, lower_expr(cx, e)?, width, "a stream operand")?;
    restream(cx, stored, &ty)
}

/// The value of a `Streaming` expression: its `W`-bit stream.
pub(crate) fn lower_streaming<'ctx, 'd>(
    cx: &LowerCtx<'ctx, 'd, '_>,
    e: sv_lang::Expression<'d>,
) -> Result<PackedValue<'ctx>, LoweringError> {
    let mut parts = Vec::new();
    for s in e.streams() {
        if let Some(w) = s.with_expr() {
            parts.push(crate::stream_with::pack(cx, s.operand(), w)?);
            continue;
        }
        parts.push(operand_stream(cx, s.operand())?);
    }
    if parts.is_empty() {
        return Err(LoweringError::unsupported_at("an empty stream", &cx.fn_sym));
    }
    let v = op(cx).concat4(&parts);
    Ok(match e.streaming_slice_size() {
        Some(n) if n > 0 => reverse_slices(cx, v, n as u32),
        _ => v,
    })
}

/// The gathered runtime stream (slice size `slice`) delivered to the
/// fixed-width type `to`: left-aligned, zero-filled on the right.
fn runtime_stream_to_fixed<'ctx>(
    cx: &LowerCtx<'ctx, '_, '_>,
    slice: i64,
    to: &Type<'_>,
    at: sv_lang::SourceLoc,
) -> Result<LoweredValue<'ctx>, LoweringError> {
    let i64t = cx.ctx.i64_type();
    let width = to.value_width();
    let words = (2 * width).div_ceil(64).max(1);
    let buf = entry_alloca(cx, op(cx).i_n(64 * words), "stream_out");
    let len = raw_call(
        cx,
        [0; 5],
        CollOp::StreamOut,
        i64t.const_zero(),
        Args {
            index: Some(i64t.const_int(slice as u64, true)),
            index2: Some(i64t.const_int(u64::from(width), false)),
            val: Some(buf),
            ..Args::default()
        },
    );
    // LRM 1800-2023 11.4.14.3: a fixed-size target narrower than the stream
    // is an error (the runtime-length stream is checked here, at run time).
    let over =
        cx.b.build_int_compare(
            inkwell::IntPredicate::UGT,
            len,
            i64t.const_int(u64::from(width), false),
            "stream_over",
        )
        .unwrap();
    crate::stmt_shared::error_if(
        cx,
        over,
        at,
        &format!("a stream longer than its {width}-bit target"),
    )?;
    let raw = crate::collections::load_words(cx, buf, words, width);
    let s = crate::collections::from_four_state(cx, raw, width, false);
    from_stream(cx, s, to)
}

/// A `StreamingConcat` conversion: the stream, left-aligned in the target
/// type's stored bits (zero-filled on the right).
pub(crate) fn lower_stream_conversion<'ctx, 'd>(
    cx: &LowerCtx<'ctx, 'd, '_>,
    e: sv_lang::Expression<'d>,
) -> Result<LoweredValue<'ctx>, LoweringError> {
    let (Some(inner), Some(to)) = (e.conversion_operand(), e.expr_type()) else {
        return Err(LoweringError::unsupported_at(
            "a malformed stream conversion",
            &cx.fn_sym,
        ));
    };
    if to.is_string() || needs_runtime(inner) {
        let slice = gather(cx, inner)?;
        if to.is_string() {
            let id = rt(cx, CollOp::StreamStr, 0, slice, 0, None);
            return Ok(LoweredValue::Str(crate::strings::StrValue { llvm: id }));
        }
        return runtime_stream_to_fixed(cx, slice, &to, e.source_range().start);
    }
    let stream = if inner.kind() == ExpressionKind::Streaming {
        lower_streaming(cx, inner)?
    } else {
        operand_stream(cx, inner)?
    };
    let width = to.value_width();
    if stream.width > width {
        return Err(LoweringError::unsupported_at(
            &format!("a {}-bit stream into a {width}-bit target", stream.width),
            &cx.fn_sym,
        ));
    }
    let op = op(cx);
    let aligned = if stream.width == width {
        stream
    } else {
        let zeros = crate::value::const_bits(
            cx.ctx,
            &vec![sv_lang::Bit::Zero; (width - stream.width) as usize],
            false,
            cx.repr,
        );
        op.concat4(&[stream, zeros])
    };
    from_stream(cx, aligned, &to)
}

/// Whether a stream operand has a runtime length.
fn dynamic(e: &sv_lang::Expression<'_>) -> bool {
    e.expr_type()
        .is_some_and(|t| t.is_string() || t.is_dynamically_sized_array())
}

/// Whether `e` is a queue or dynamic array: `string'(e)` is a bit-stream
/// cast of its elements' bytes (LRM 1800-2023 6.24.3), not a handle read.
pub(crate) fn is_collection_operand(e: sv_lang::Expression<'_>) -> bool {
    e.kind() != ExpressionKind::Streaming
        && e.expr_type()
            .is_some_and(|t| !t.is_string() && t.is_dynamically_sized_array())
}

/// Whether `s` is `array with [...]` on a fixed-size array whose element
/// count is a run-time value (LRM 1800-2023 11.4.14.4).
fn runtime_fixed_with(s: &sv_lang::StreamExpr<'_>) -> bool {
    s.with_expr().is_some_and(|w| {
        !dynamic(&s.operand()) && crate::stream_with::runtime_count(s.operand(), w)
    })
}

/// Whether the stream `inner` (a `Streaming` expression or a lone operand)
/// has a runtime length.
fn needs_runtime(inner: sv_lang::Expression<'_>) -> bool {
    if inner.kind() == ExpressionKind::Streaming {
        inner
            .streams()
            .iter()
            .any(|s| runtime_fixed_with(s) || needs_runtime(s.operand()))
    } else {
        dynamic(&inner)
    }
}

/// `sv_rt_coll(op, h, {index, index2, val})` for a stream-buffer operation.
fn rt<'ctx>(
    cx: &LowerCtx<'ctx, '_, '_>,
    what: CollOp,
    h: u64,
    index: i64,
    index2: i64,
    val: Option<inkwell::values::PointerValue<'ctx>>,
) -> inkwell::values::IntValue<'ctx> {
    let i64t = cx.ctx.i64_type();
    raw_call(
        cx,
        [0; 5],
        what,
        i64t.const_int(h, false),
        Args {
            index: Some(i64t.const_int(index as u64, true)),
            index2: Some(i64t.const_int(index2 as u64, true)),
            val,
            ..Args::default()
        },
    )
}

/// Appends the fixed-width bits `v` to the runtime stream buffer.
pub(crate) fn push_bits<'ctx>(cx: &LowerCtx<'ctx, '_, '_>, v: PackedValue<'ctx>) {
    let words = (2 * v.width).div_ceil(64).max(1);
    let buf = crate::collections::words_buf(
        cx,
        words,
        crate::collections::four_state(cx, v),
        "stream_bits",
    );
    rt(cx, CollOp::StreamPush, 0, i64::from(v.width), 0, Some(buf));
}

/// Appends operand `e`'s bit-stream to the runtime stream buffer.
fn push_operand<'ctx, 'd>(
    cx: &LowerCtx<'ctx, 'd, '_>,
    e: sv_lang::Expression<'d>,
) -> Result<(), LoweringError> {
    let i64t = cx.ctx.i64_type();
    if let Some(c) = crate::collections::coll_of_expr(&e) {
        let struct_elems = !c.assoc && vj_sim::state::storage::is_unpacked_aggregate(&c.elem);
        if !c.assoc && (c.elem.is_string() || struct_elems) {
            // LRM 11.4.14: a queue or dynamic array of strings streams its
            // elements in order, each string as its bytes; of unpacked
            // structs / unions (a bit-stream type, 6.24.3), each element's
            // members in declaration order (`restream`).
            let h = crate::collections::handle_of(cx, e)?;
            let n = crate::collections::size_of_handle(cx, &c, h);
            let f =
                cx.b.get_insert_block()
                    .and_then(|bb| bb.get_parent())
                    .expect("lowering happens inside a function");
            let slot = entry_alloca(cx, i64t, "stream_str_i");
            cx.b.build_store(slot, i64t.const_zero()).unwrap();
            let head = cx.ctx.append_basic_block(f, "stream_str_head");
            let body = cx.ctx.append_basic_block(f, "stream_str_body");
            let done = cx.ctx.append_basic_block(f, "stream_str_done");
            cx.b.build_unconditional_branch(head).unwrap();
            cx.b.position_at_end(head);
            let i = cx.b.build_load(i64t, slot, "i").unwrap().into_int_value();
            let more =
                cx.b.build_int_compare(inkwell::IntPredicate::SLT, i, n, "more")
                    .unwrap();
            cx.b.build_conditional_branch(more, body, done).unwrap();
            cx.b.position_at_end(body);
            let el = crate::collections::read_at_coll(cx, &c, h, i);
            let LoweredValue::Packed(id) = el else {
                return Err(LoweringError::unsupported_at(
                    "a string or unpacked-struct queue element that is not a packed value",
                    &cx.fn_sym,
                ));
            };
            if struct_elems {
                let v = restream(cx, id, &c.elem)?;
                push_bits(cx, v);
            } else {
                // A string element is stored as its runtime string id.
                let id = crate::collections::op(cx).lane_val(id);
                raw_call(cx, [0; 5], CollOp::StreamPushStr, id, Args::default());
            }
            let next =
                cx.b.build_int_add(i, i64t.const_int(1, false), "i_next")
                    .unwrap();
            cx.b.build_store(slot, next).unwrap();
            cx.b.build_unconditional_branch(head).unwrap();
            cx.b.position_at_end(done);
            return Ok(());
        }
        if c.assoc
            || !c.elem.is_integral()
            || vj_sim::state::storage::is_unpacked_aggregate(&c.elem)
        {
            return Err(LoweringError::unsupported_at(
                "a streamed queue or dynamic array of non-integral elements",
                &cx.fn_sym,
            ));
        }
        let h = crate::collections::handle_of(cx, e)?;
        raw_call(
            cx,
            [0; 5],
            CollOp::StreamPushColl,
            h,
            Args {
                index: Some(i64t.const_int(u64::from(c.width), false)),
                ..Args::default()
            },
        );
        return Ok(());
    }
    if e.expr_type().is_some_and(|t| t.is_string()) {
        let s = crate::strings::to_string(cx, lower_expr_h(cx, e)?)?;
        raw_call(cx, [0; 5], CollOp::StreamPushStr, s.llvm, Args::default());
        return Ok(());
    }
    if e.kind() == ExpressionKind::Streaming && needs_runtime(e) {
        // LRM 1800-2023 11.4.14.4: a nested stream is re-ordered by its own
        // slice size before it joins the enclosing stream, so it gathers in a
        // buffer of its own.
        rt(cx, CollOp::StreamEnter, 0, 0, 0, None);
        push_streams(cx, e)?;
        let slice = e
            .streaming_slice_size()
            .filter(|n| *n > 0)
            .map_or(0, |n| n as i64);
        rt(cx, CollOp::StreamLeave, 0, slice, 0, None);
        return Ok(());
    }
    let v = operand_stream(cx, e)?;
    push_bits(cx, v);
    Ok(())
}

/// Appends the operands of the `Streaming` expression `inner` to the runtime
/// stream buffer, in order.
fn push_streams<'ctx, 'd>(
    cx: &LowerCtx<'ctx, 'd, '_>,
    inner: sv_lang::Expression<'d>,
) -> Result<(), LoweringError> {
    for s in inner.streams() {
        if let Some(w) = s.with_expr() {
            if crate::collections::coll_of_expr(&s.operand()).is_some() {
                crate::stream_with::push_collection_range(cx, s.operand(), w)?;
            } else if runtime_fixed_with(&s) {
                crate::stream_with::push_fixed_range(cx, s.operand(), w)?;
            } else {
                let v = crate::stream_with::pack(cx, s.operand(), w)?;
                push_bits(cx, v);
            }
            continue;
        }
        push_operand(cx, s.operand())?;
    }
    Ok(())
}

/// Gathers the stream `inner` (a `Streaming` expression or a lone operand)
/// in the runtime stream buffer; returns its slice size (0: `{>>...}`).
fn gather<'ctx, 'd>(
    cx: &LowerCtx<'ctx, 'd, '_>,
    inner: sv_lang::Expression<'d>,
) -> Result<i64, LoweringError> {
    rt(cx, CollOp::StreamClear, 0, 0, 0, None);
    if inner.kind() != ExpressionKind::Streaming {
        push_operand(cx, inner)?;
        return Ok(0);
    }
    push_streams(cx, inner)?;
    Ok(inner
        .streaming_slice_size()
        .filter(|n| *n > 0)
        .map_or(0, |n| n as i64))
}

/// The stream `inner` as a string of its bytes (the last one zero-filled).
pub(crate) fn stream_to_string<'ctx, 'd>(
    cx: &LowerCtx<'ctx, 'd, '_>,
    inner: sv_lang::Expression<'d>,
) -> Result<crate::strings::StrValue<'ctx>, LoweringError> {
    let slice = gather(cx, inner)?;
    Ok(crate::strings::StrValue {
        llvm: rt(cx, CollOp::StreamStr, 0, slice, 0, None),
    })
}

/// The stream conversion `e` into a queue or dynamic array of shape `c`: a
/// new collection of its bits, cut into elements (the last one
/// zero-filled).
pub(crate) fn stream_to_collection<'ctx, 'd>(
    cx: &LowerCtx<'ctx, 'd, '_>,
    e: sv_lang::Expression<'d>,
    c: &Coll<'d>,
) -> Result<inkwell::values::IntValue<'ctx>, LoweringError> {
    let inner = e.conversion_operand().ok_or_else(|| {
        LoweringError::unsupported_at("a malformed stream conversion", &cx.fn_sym)
    })?;
    if c.assoc || !c.elem.is_integral() || vj_sim::state::storage::is_unpacked_aggregate(&c.elem) {
        return Err(LoweringError::unsupported_at(
            "a stream into an associative array or an array of non-integral elements",
            &cx.fn_sym,
        ));
    }
    let slice = gather(cx, inner)?;
    let i64t = cx.ctx.i64_type();
    Ok(crate::collections::call(
        cx,
        c,
        CollOp::StreamColl,
        i64t.const_zero(),
        Args {
            index: Some(i64t.const_int(slice as u64, true)),
            index2: Some(i64t.const_int(u64::from(c.width), false)),
            ..Args::default()
        },
    ))
}

/// A bit-stream cast `T'(v)` between two types of the same bit-stream
/// width: the operand's stream, read back as `T`.
pub(crate) fn lower_bitstream_cast<'ctx, 'd>(
    cx: &LowerCtx<'ctx, 'd, '_>,
    e: sv_lang::Expression<'d>,
) -> Result<LoweredValue<'ctx>, LoweringError> {
    let (Some(inner), Some(to)) = (e.conversion_operand(), e.expr_type()) else {
        return Err(LoweringError::unsupported_at(
            "a malformed bit-stream cast",
            &cx.fn_sym,
        ));
    };
    if needs_runtime(inner) && !to.is_string() {
        // LRM 11.4.14.4: a runtime-length stream cast; its length is
        // checked by the runtime when delivered.
        let slice = gather(cx, inner)?;
        return runtime_stream_to_fixed(cx, slice, &to, e.source_range().start);
    }
    let stream = if inner.kind() == ExpressionKind::Streaming {
        lower_streaming(cx, inner)?
    } else {
        operand_stream(cx, inner)?
    };
    if stream.width != to.value_width() {
        return Err(LoweringError::unsupported_at(
            &format!(
                "a bit-stream cast of {} bits to a {}-bit type",
                stream.width,
                to.value_width()
            ),
            &cx.fn_sym,
        ));
    }
    from_stream(cx, stream, &to)
}

/// The value of type `to` whose bit-stream is `s`.
fn from_stream<'ctx>(
    cx: &LowerCtx<'ctx, '_, '_>,
    s: PackedValue<'ctx>,
    to: &Type<'_>,
) -> Result<LoweredValue<'ctx>, LoweringError> {
    let stored = unstream(cx, s, to)?;
    Ok(if to.is_floating() {
        LoweredValue::Real(crate::real::bits_to_real(cx, stored))
    } else {
        LoweredValue::Packed(PackedValue {
            signed: to.is_signed(),
            ..stored
        })
    })
}

/// The inverse of [`restream`]: a type's stored bits from its bit-stream.
pub(crate) fn unstream<'ctx>(
    cx: &LowerCtx<'ctx, '_, '_>,
    s: PackedValue<'ctx>,
    ty: &Type<'_>,
) -> Result<PackedValue<'ctx>, LoweringError> {
    use vj_sim::state::storage;
    let bad = || LoweringError::unsupported_at("a streamed value's shape", &cx.fn_sym);
    if ty.is_unpacked_array() && !ty.is_dynamically_sized_array() {
        let range = ty.fixed_unpacked_array_range();
        let elem = ty.element_type().ok_or_else(bad)?;
        let ew = storage::storage_width(&elem).ok_or_else(bad)? as u32;
        let n = range.width();
        // Element `i` in stream order sits `i` elements below the top.
        let mut stored = vec![None; n as usize];
        for i in 0..n {
            let piece = bits(cx, s, (n - 1 - i) * ew, ew);
            let pos = if range.left <= range.right {
                i
            } else {
                n - 1 - i
            };
            stored[pos as usize] = Some(unstream(cx, piece, &elem)?);
        }
        // Stored lowest index least significant: concatenate highest first.
        let parts: Vec<_> = stored
            .into_iter()
            .rev()
            .map(|p| p.expect("every position"))
            .collect();
        return Ok(op(cx).concat4(&parts));
    }
    if storage::is_unpacked_aggregate(ty) && !ty.is_unpacked_union() {
        let fields: Vec<_> = ty.fields().collect();
        let mut parts = Vec::with_capacity(fields.len());
        let mut top = s.width;
        for f in &fields {
            let fty = f.value_type().ok_or_else(bad)?;
            let w = storage::storage_width(&fty).ok_or_else(bad)? as u32;
            top -= w;
            parts.push(unstream(cx, bits(cx, s, top, w), &fty)?);
        }
        return Ok(op(cx).concat4(&parts));
    }
    Ok(s)
}

/// The bit-stream width of a fixed-size stream target of type `ty`: a fixed
/// unpacked array or unpacked struct streams as its elements/members
/// concatenated (LRM 1800-2023 11.4.14.1), which is its stored width.
fn target_width(cx: &LowerCtx<'_, '_, '_>, ty: &Type<'_>) -> Result<u32, LoweringError> {
    use vj_sim::state::storage;
    if ty.is_unpacked_array() {
        return Ok(storage::storage_width(ty).ok_or_else(|| {
            LoweringError::unsupported_at("an unpacked-array stream target's shape", &cx.fn_sym)
        })? as u32);
    }
    Ok(ty.value_width())
}

/// The fixed bit width of the stream target `e` when it is a nested stream
/// (`{>>{ {<<4{a}}, b }} = x`, LRM 1800-2023 11.4.14.4): the sum of its
/// operands', and whether one of them (a queue, dynamic array or string, at
/// any depth; only one may) takes whatever the rest leave.
fn nested_target_parts(
    cx: &LowerCtx<'_, '_, '_>,
    e: sv_lang::Expression<'_>,
) -> Result<(u32, bool), LoweringError> {
    let mut total = 0;
    let mut open = false;
    for s in e.streams() {
        let operand = s.operand();
        let mut has_open = false;
        total += if let Some(w) = s.with_expr() {
            crate::stream_with::width(cx, operand, w)?.ok_or_else(|| {
                LoweringError::unsupported_at(
                    "a nested stream target with a `with` range on a queue or dynamic array",
                    &cx.fn_sym,
                )
            })?
        } else if operand.kind() == ExpressionKind::Streaming {
            let (w, o) = nested_target_parts(cx, operand)?;
            has_open = o;
            w
        } else if dynamic(&operand) {
            has_open = true;
            0
        } else {
            let ty = operand.expr_type().ok_or_else(|| {
                LoweringError::unsupported_at("a stream target has no type", &cx.fn_sym)
            })?;
            target_width(cx, &ty)?
        };
        // Only the first open operand is greedy; the others stay empty
        // (LRM 1800-2023 11.4.14.4), which unpacking the nested target
        // itself arranges.
        open |= has_open;
    }
    Ok((total, open))
}

/// Unpacks `value` into the `Streaming` lvalue `lhs`: the source's leading
/// (most significant) `W` bits, un-reordered, split across the operands
/// most significant first. Each operand's bits (in stored form) go to
/// `assign(operand, bits)`.
pub(crate) fn unpack_stream<'ctx, 'd>(
    cx: &LowerCtx<'ctx, 'd, '_>,
    lhs: sv_lang::Expression<'d>,
    value: PackedValue<'ctx>,
    assign: &mut Assign<'ctx, 'd, '_>,
    mut sink: Option<&mut CollSink<'ctx, 'd, '_>>,
) -> Result<(), LoweringError> {
    let streams = lhs.streams();
    let mut widths = Vec::with_capacity(streams.len());
    // Queue / dynamic array / string operands after the open one: LRM
    // 1800-2023 11.4.14.4 leaves them empty.
    let mut empty = vec![false; streams.len()];
    // A queue, dynamic array or string operand takes whatever the others
    // leave (only one may).
    let mut open = None;
    for (i, s) in streams.iter().enumerate() {
        if let Some(w) = s.with_expr() {
            // A fixed-size array takes the selected elements' bits.
            let Some(bits) = crate::stream_with::width(cx, s.operand(), w)? else {
                return Err(LoweringError::unsupported_at(
                    "a stream target `with [...]` range on an operand that is not a \
                     fixed-size array in a context other than a blocking assignment (nonblocking, &c.)",
                    &cx.fn_sym,
                ));
            };
            widths.push(bits);
            continue;
        }
        let operand = s.operand();
        if operand.kind() == ExpressionKind::Streaming {
            // A nested stream holding the open operand is the open one: it
            // takes what the others leave and splits that among its own.
            let (w, nested_open) = nested_target_parts(cx, operand)?;
            if nested_open && open.is_none() {
                open = Some(i);
            }
            widths.push(w);
            continue;
        }
        let ty = operand.expr_type().ok_or_else(|| {
            LoweringError::unsupported_at("a stream target has no type", &cx.fn_sym)
        })?;
        if dynamic(&operand) {
            if open.is_none() {
                open = Some(i);
            } else {
                empty[i] = true;
            }
            widths.push(0);
            continue;
        }
        widths.push(target_width(cx, &ty)?);
    }
    if let Some(i) = open {
        let fixed: u32 = widths.iter().sum();
        let extra = value.width.checked_sub(fixed).ok_or_else(|| {
            LoweringError::unsupported_at(
                &format!(
                    "a {}-bit source unpacked into a {fixed}-bit stream",
                    value.width
                ),
                &cx.fn_sym,
            )
        })?;
        widths[i] += extra;
    }
    let total: u32 = widths.iter().sum();
    if value.width < total {
        return Err(LoweringError::unsupported_at(
            &format!(
                "a {}-bit source unpacked into a {total}-bit stream",
                value.width
            ),
            &cx.fn_sym,
        ));
    }
    let lead = bits(cx, value, value.width - total, total);
    let v = match lhs.streaming_slice_size() {
        Some(n) if n > 0 => unreverse_slices(cx, lead, n as u32),
        _ => lead,
    };
    let mut top = total;
    for (i, (s, w)) in streams.iter().zip(widths).enumerate() {
        top -= w;
        let operand = s.operand();
        if (open == Some(i) && operand.kind() != ExpressionKind::Streaming) || empty[i] {
            let piece = (w > 0).then(|| bits(cx, v, top, w));
            assign_dynamic(cx, operand, piece, sink.as_deref_mut())?;
            continue;
        }
        if let Some(wx) = s.with_expr() {
            assign(operand, Some(wx), bits(cx, v, top, w))?;
            continue;
        }
        if operand.kind() == ExpressionKind::Streaming {
            // A nested stream re-orders its own bits by its own slice size.
            unpack_stream(
                cx,
                operand,
                bits(cx, v, top, w),
                assign,
                sink.as_deref_mut(),
            )?;
            continue;
        }
        let ty = operand.expr_type().expect("checked above");
        let piece = unstream(cx, bits(cx, v, top, w), &ty)?;
        assign(operand, None, piece)?;
    }
    Ok(())
}

/// Stores the bits of one stream target operand: the operand, its `with`
/// range if any (then the bits are the selected elements in stream form,
/// see [`crate::stream_with::unpack`]), and the bits.
pub(crate) type Assign<'ctx, 'd, 'a> = dyn FnMut(
        sv_lang::Expression<'d>,
        Option<sv_lang::Expression<'d>>,
        PackedValue<'ctx>,
    ) -> Result<(), LoweringError>
    + 'a;

/// Stores `piece` (`None`: no bits) into the queue, dynamic array or
/// string stream target `operand`.
fn assign_dynamic<'ctx, 'd>(
    cx: &LowerCtx<'ctx, 'd, '_>,
    operand: sv_lang::Expression<'d>,
    piece: Option<PackedValue<'ctx>>,
    sink: Option<&mut CollSink<'ctx, 'd, '_>>,
) -> Result<(), LoweringError> {
    rt(cx, CollOp::StreamClear, 0, 0, 0, None);
    if let Some(p) = piece {
        push_bits(cx, p);
    }
    store_stream_buffer(cx, operand, sink)
}

/// Receives the new queue / dynamic array handle (owned) a stream unpack
/// built for `operand` in place of the blocking release-and-store: a
/// nonblocking assignment stages it (LRM 1800-2023 10.4.2).
pub(crate) type CollSink<'ctx, 'd, 'a> =
    dyn FnMut(sv_lang::Expression<'d>, IntValue<'ctx>) -> Result<(), LoweringError> + 'a;

/// Takes the whole stream buffer into the queue, dynamic array or string
/// target `operand` (`sink`: hand a collection's new handle over instead of
/// storing it now; a string target stages its new string the same way).
fn store_stream_buffer<'ctx, 'd>(
    cx: &LowerCtx<'ctx, 'd, '_>,
    operand: sv_lang::Expression<'d>,
    sink: Option<&mut CollSink<'ctx, 'd, '_>>,
) -> Result<(), LoweringError> {
    if let Some(c) = crate::collections::coll_of_expr(&operand) {
        let struct_elems = !c.assoc && vj_sim::state::storage::is_unpacked_aggregate(&c.elem);
        if c.assoc || !(c.elem.is_integral() || struct_elems) {
            return Err(LoweringError::unsupported_at(
                "a stream target array of non-integral elements",
                &cx.fn_sym,
            ));
        }
        let i64t = cx.ctx.i64_type();
        let new = crate::collections::call(
            cx,
            &c,
            CollOp::StreamColl,
            i64t.const_zero(),
            Args {
                index2: Some(i64t.const_int(u64::from(c.width), false)),
                ..Args::default()
            },
        );
        if struct_elems {
            unstream_elements(cx, &c, new)?;
        }
        if let Some(sink) = sink {
            return sink(operand, new);
        }
        let old = crate::collections::handle_of(cx, operand)?;
        crate::collections::call(cx, &c, CollOp::Release, old, Args::default());
        crate::collections::set_handle(cx, operand, new)?;
        return Ok(());
    }
    let id = rt(cx, CollOp::StreamStr, 0, 0, 0, None);
    if let Some(sink) = sink {
        // A string is a handle too (`strings::as_packed` is its 64-bit id):
        // the new string is staged for the NBA region like a collection's.
        return sink(operand, id);
    }
    crate::expr::store_to(
        cx,
        operand,
        LoweredValue::Str(crate::strings::StrValue { llvm: id }),
    )?;
    Ok(())
}

/// Converts every element of the new queue / dynamic array `h` (of unpacked
/// struct or union elements, holding the elements' bit-streams) to the
/// elements' stored form ([`unstream`], LRM 1800-2023 11.4.14.4).
fn unstream_elements<'ctx>(
    cx: &LowerCtx<'ctx, '_, '_>,
    c: &Coll<'_>,
    h: inkwell::values::IntValue<'ctx>,
) -> Result<(), LoweringError> {
    let i64t = cx.ctx.i64_type();
    let n = crate::collections::size_of_handle(cx, c, h);
    let f =
        cx.b.get_insert_block()
            .and_then(|bb| bb.get_parent())
            .expect("lowering happens inside a function");
    let slot = entry_alloca(cx, i64t, "unstream_i");
    cx.b.build_store(slot, i64t.const_zero()).unwrap();
    let head = cx.ctx.append_basic_block(f, "unstream_head");
    let body = cx.ctx.append_basic_block(f, "unstream_body");
    let done = cx.ctx.append_basic_block(f, "unstream_done");
    cx.b.build_unconditional_branch(head).unwrap();
    cx.b.position_at_end(head);
    let i = cx.b.build_load(i64t, slot, "i").unwrap().into_int_value();
    let more =
        cx.b.build_int_compare(inkwell::IntPredicate::SLT, i, n, "more")
            .unwrap();
    cx.b.build_conditional_branch(more, body, done).unwrap();
    cx.b.position_at_end(body);
    let LoweredValue::Packed(el) = crate::collections::read_at_coll(cx, c, h, i) else {
        return Err(LoweringError::unsupported_at(
            "a struct queue element that is not a packed value",
            &cx.fn_sym,
        ));
    };
    let stored = unstream(cx, el, &c.elem)?;
    let val = crate::collections::elem_buf(cx, c, LoweredValue::Packed(stored))?;
    crate::collections::call(
        cx,
        c,
        CollOp::Set,
        h,
        Args {
            index: Some(i),
            index2: Some(i64t.const_zero()),
            val: Some(val),
            ..Args::default()
        },
    );
    let next =
        cx.b.build_int_add(i, i64t.const_int(1, false), "i_next")
            .unwrap();
    cx.b.build_store(slot, next).unwrap();
    cx.b.build_unconditional_branch(head).unwrap();
    cx.b.position_at_end(done);
    Ok(())
}

/// Whether the source `rhs` of a stream-target assignment has a runtime
/// length (a queue, dynamic array or string, or a stream of one), so its
/// width is not known when the target is lowered.
pub(crate) fn source_needs_runtime(rhs: sv_lang::Expression<'_>) -> bool {
    needs_runtime(rhs)
}

/// Whether the stream target `lhs` has a `with [...]` operand on a queue or
/// dynamic array: its element count is known only at run time (LRM
/// 1800-2023 11.4.14.4), so even a fixed-width source unpacks through the
/// runtime stream buffer ([`unpack_runtime_source`]).
pub(crate) fn target_needs_runtime(lhs: sv_lang::Expression<'_>) -> bool {
    lhs.kind() == ExpressionKind::Streaming
        && lhs
            .streams()
            .iter()
            .any(|s| (s.with_expr().is_some() && dynamic(&s.operand())) || runtime_fixed_with(s))
}

/// Whether `{...} <= rhs` unpacks through the runtime stream buffer: the
/// source has a runtime length, or a target operand is a queue, dynamic
/// array or string (whose length the stream decides).
pub(crate) fn nonblocking_needs_runtime(
    lhs: sv_lang::Expression<'_>,
    rhs: sv_lang::Expression<'_>,
) -> bool {
    lhs.kind() == ExpressionKind::Streaming
        && (needs_runtime(rhs) || target_has_dynamic(lhs) || target_needs_runtime(lhs))
}

/// Whether a stream target holds a queue, dynamic array or string operand,
/// at any nesting depth.
fn target_has_dynamic(e: sv_lang::Expression<'_>) -> bool {
    e.streams().iter().any(|s| {
        let o = s.operand();
        if o.kind() == ExpressionKind::Streaming {
            target_has_dynamic(o)
        } else {
            dynamic(&o)
        }
    })
}

/// `{<<n{a, b, ...}} = rhs` for a source of runtime length (a queue,
/// dynamic array or string; LRM 1800-2023 §11.4.14.3/.4): the source's bits
/// gather in the runtime stream buffer, which is sliced as the target asks
/// and read operand by operand at offsets computed at run time. Each fixed
/// operand's bits (stored form) go to `assign(operand, bits)`; a queue,
/// dynamic array or string operand takes whatever the others leave. A source
/// shorter than the targets leaves the missing bits zero.
pub(crate) fn unpack_runtime_source<'ctx, 'd>(
    cx: &LowerCtx<'ctx, 'd, '_>,
    lhs: sv_lang::Expression<'d>,
    rhs: sv_lang::Expression<'d>,
    assign: &mut Assign<'ctx, 'd, '_>,
    sink: Option<&mut CollSink<'ctx, 'd, '_>>,
    rows: Option<&mut crate::stream_with::RowSink<'ctx, '_>>,
) -> Result<(), LoweringError> {
    // The source, in the order a `{>>{...}}` of it would give.
    let rhs_slice = gather(cx, rhs)?;
    if rhs_slice > 0 {
        rt(cx, CollOp::StreamReslice, 0, rhs_slice, 0, None);
    }
    unpack_buffer(cx, lhs, assign, sink, rows)
}

/// [`unpack_runtime_source`] once the source's bits are in the run-time
/// stream buffer. A nested target holding the open operand unpacks a range
/// of the buffer through this again (LRM 1800-2023 11.4.14.4).
fn unpack_buffer<'ctx, 'd>(
    cx: &LowerCtx<'ctx, 'd, '_>,
    lhs: sv_lang::Expression<'d>,
    assign: &mut Assign<'ctx, 'd, '_>,
    mut sink: Option<&mut CollSink<'ctx, 'd, '_>>,
    mut rows: Option<&mut crate::stream_with::RowSink<'ctx, '_>>,
) -> Result<(), LoweringError> {
    let streams = lhs.streams();
    let mut widths: Vec<u64> = Vec::with_capacity(streams.len());
    let mut open = None;
    // Queue / dynamic array / string operands after the open one stay empty.
    let mut empties: Vec<usize> = Vec::new();
    let mut withs: Vec<usize> = Vec::new();
    for (i, s) in streams.iter().enumerate() {
        if let Some(w) = s.with_expr() {
            // A fixed-size array operand: a constant number of elements.
            if !runtime_fixed_with(s)
                && let Some(bits) = crate::stream_with::width(cx, s.operand(), w)?
            {
                widths.push(u64::from(bits));
                continue;
            }
            withs.push(i);
            widths.push(0);
            continue;
        }
        let operand = s.operand();
        if operand.kind() == ExpressionKind::Streaming {
            let (w, nested_open) = nested_target_parts(cx, operand)?;
            if nested_open && open.is_none() {
                open = Some(i);
            }
            widths.push(u64::from(w));
            continue;
        }
        let ty = operand.expr_type().ok_or_else(|| {
            LoweringError::unsupported_at("a stream target has no type", &cx.fn_sym)
        })?;
        if dynamic(&operand) {
            if open.is_none() {
                open = Some(i);
            } else {
                empties.push(i);
            }
            widths.push(0);
            continue;
        }
        widths.push(u64::from(target_width(cx, &ty)?));
    }
    let i64t = cx.ctx.i64_type();
    let fixed: u64 = widths.iter().sum();
    // A `with [0 +: n]` operand takes `n` elements, a width known only at
    // run time (LRM 1800-2023 11.4.14.4).
    // Its width is evaluated once the operands before it are assigned (an
    // earlier operand may be the count), so no leading-bits limit applies.
    // The target's own slicing; without an open operand only the leading
    // `fixed` bits take part.
    let slice = lhs
        .streaming_slice_size()
        .filter(|n| *n > 0)
        .map_or(0, |n| n as i64);
    let limit = if open.is_none() && withs.is_empty() {
        fixed as i64
    } else {
        0
    };
    let len = rt(cx, CollOp::StreamReslice, 0, slice, limit, None);
    if withs.is_empty() {
        // LRM 1800-2023 11.4.14.3: a source with fewer bits than the
        // fixed-size targets is an error (`{>>{a, b, c}} = 23'b1`).
        let short =
            cx.b.build_int_compare(
                inkwell::IntPredicate::ULT,
                len,
                i64t.const_int(fixed, false),
                "stream_short",
            )
            .unwrap();
        crate::stmt_shared::error_if(
            cx,
            short,
            lhs.source_range().start,
            &format!("a stream with fewer than the {fixed} bits its targets need"),
        )?;
    }
    // Operand offsets: `cur` (run-time, as a `with` operand's width is) up to
    // the open operand, counted back from the end of the stream after it.
    let n = streams.len();
    let zero = i64t.const_zero();
    let konst = |v: u64| i64t.const_int(v, false);
    let mut cur = zero;
    let mut with_w: Vec<Option<(IntValue<'ctx>, IntValue<'ctx>)>> = vec![None; n];
    // `suffix[i]`: the bits of operands `i..` (only after the open operand).
    let mut suffix = vec![zero; n + 1];
    for (i, (s, w)) in streams.iter().zip(&widths).enumerate() {
        if empties.contains(&i) {
            consume_range(cx, zero, zero, None, s.operand(), sink.as_deref_mut())?;
            continue;
        }
        if open == Some(i) && s.operand().kind() == ExpressionKind::Streaming {
            // A nested target holding the open operand: it unpacks the
            // range the operands around it leave, as a stream of its own.
            for j in i + 1..n {
                if withs.contains(&j) {
                    with_w[j] = Some(with_width(cx, &streams[j])?);
                }
            }
            for j in (i + 1..n).rev() {
                let bits = with_w[j].map_or_else(|| konst(widths[j]), |(b, _)| b);
                suffix[j] =
                    cx.b.build_int_add(suffix[j + 1], bits, "stream_tail")
                        .unwrap();
            }
            let end =
                cx.b.build_int_sub(len, suffix[i + 1], "stream_end")
                    .unwrap();
            rt(cx, CollOp::StreamEnter, 0, 1, 0, None);
            raw_call(
                cx,
                [0; 5],
                CollOp::StreamRange,
                i64t.const_zero(),
                Args {
                    index: Some(cur),
                    index2: Some(end),
                    ..Args::default()
                },
            );
            unpack_buffer(
                cx,
                s.operand(),
                assign,
                sink.as_deref_mut(),
                rows.as_deref_mut(),
            )?;
            rt(cx, CollOp::StreamLeave, 0, 0, 1, None);
            continue;
        }
        if open == Some(i) {
            // The operands after it take their bits from the end of the
            // stream: their widths (a `with` operand's included) are needed
            // first.
            for j in i + 1..n {
                if withs.contains(&j) {
                    with_w[j] = Some(with_width(cx, &streams[j])?);
                }
            }
            for j in (i + 1..n).rev() {
                let bits = with_w[j].map_or_else(|| konst(widths[j]), |(b, _)| b);
                suffix[j] =
                    cx.b.build_int_add(suffix[j + 1], bits, "stream_tail")
                        .unwrap();
            }
            let end =
                cx.b.build_int_sub(len, suffix[i + 1], "stream_end")
                    .unwrap();
            consume_range(
                cx,
                cur,
                end,
                None,
                streams[i].operand(),
                sink.as_deref_mut(),
            )?;
            continue;
        }
        let after_open = open.is_some_and(|o| i > o);
        let start = if after_open {
            cx.b.build_int_sub(len, suffix[i], "stream_off").unwrap()
        } else {
            cur
        };
        if withs.contains(&i) {
            // A `with [a +: n]` operand takes `n` elements, a width known
            // only at run time (LRM 1800-2023 11.4.14.4); evaluated once the
            // operands before it are assigned (an earlier one may be the
            // count).
            let (bits, pad) = match with_w[i] {
                Some(v) => v,
                None => with_width(cx, s)?,
            };
            if runtime_fixed_with(s) {
                if sink.is_some() && rows.is_none() {
                    return Err(LoweringError::unsupported_at(
                        "a nonblocking stream unpack into a fixed-size array `with` operand",
                        &cx.fn_sym,
                    ));
                }
                unpack_fixed_with(
                    cx,
                    s.operand(),
                    s.with_expr().expect("checked"),
                    start,
                    rows.as_deref_mut(),
                )?;
                if !after_open {
                    cur = cx.b.build_int_add(cur, bits, "stream_cur").unwrap();
                }
                continue;
            }
            let end = cx.b.build_int_add(start, bits, "stream_end").unwrap();
            consume_range(
                cx,
                start,
                end,
                Some((pad, bits)),
                s.operand(),
                sink.as_deref_mut(),
            )?;
            if !after_open {
                cur = cx.b.build_int_add(cur, bits, "stream_cur").unwrap();
            }
            continue;
        }
        let operand = s.operand();
        let width = *w as u32;
        let words = (2 * width).div_ceil(64).max(1);
        let buf = entry_alloca(cx, op(cx).i_n(64 * words), "stream_fld");
        raw_call(
            cx,
            [0; 5],
            CollOp::StreamField,
            i64t.const_zero(),
            Args {
                index: Some(start),
                index2: Some(i64t.const_int(*w, false)),
                val: Some(buf),
                ..Args::default()
            },
        );
        let raw = crate::collections::load_words(cx, buf, words, width);
        let piece = crate::collections::from_four_state(cx, raw, width, false);
        match s.with_expr() {
            Some(wx) => assign(operand, Some(wx), piece)?,
            None if operand.kind() == ExpressionKind::Streaming => {
                // A nested stream re-orders its own bits by its own slice size.
                // One holding an operand left empty rebuilds it from the stream
                // buffer, so the enclosing stream is set aside meanwhile.
                let (_, holds_open) = nested_target_parts(cx, operand)?;
                if holds_open {
                    rt(cx, CollOp::StreamEnter, 0, 1, 0, None);
                }
                unpack_stream(cx, operand, piece, assign, sink.as_deref_mut())?;
                if holds_open {
                    rt(cx, CollOp::StreamLeave, 0, 0, 1, None);
                }
            }
            None => {
                let ty = operand.expr_type().expect("checked above");
                assign(operand, None, unstream(cx, piece, &ty)?)?;
            }
        }
        if !after_open {
            cur = cx.b.build_int_add(cur, konst(*w), "stream_cur").unwrap();
        }
    }
    Ok(())
}

/// Unpacks the run-time stream buffer, from bit `start`, into the elements
/// of the fixed-size array `operand with [...]` whose count is a run-time
/// value (LRM 1800-2023 11.4.14.4); elements outside the array are skipped
/// and reported as an error.
fn unpack_fixed_with<'ctx, 'd>(
    cx: &LowerCtx<'ctx, 'd, '_>,
    operand: sv_lang::Expression<'d>,
    w: sv_lang::Expression<'d>,
    start: IntValue<'ctx>,
    mut put: Option<&mut crate::stream_with::RowSink<'ctx, '_>>,
) -> Result<(), LoweringError> {
    let i64t = cx.ctx.i64_type();
    let zero = i64t.const_zero();
    crate::stream_with::for_each_selected(cx, operand, w, |row, p, elem| {
        let ew = elem.value_width();
        let words = (2 * ew).div_ceil(64).max(1);
        let buf = entry_alloca(cx, op(cx).i_n(64 * words), "stream_fld");
        let off =
            cx.b.build_int_mul(p, i64t.const_int(u64::from(ew), false), "with_off")
                .unwrap();
        let off = cx.b.build_int_add(start, off, "with_at_bit").unwrap();
        raw_call(
            cx,
            [0; 5],
            CollOp::StreamField,
            i64t.const_zero(),
            Args {
                index: Some(off),
                index2: Some(i64t.const_int(u64::from(ew), false)),
                val: Some(buf),
                ..Args::default()
            },
        );
        let raw = crate::collections::load_words(cx, buf, words, ew);
        let piece = crate::collections::from_four_state(cx, raw, ew, false);
        let stored = unstream(cx, piece, elem)?;
        let stored = PackedValue {
            signed: false,
            ..stored
        };
        match put.as_deref_mut() {
            Some(put) => put(row, stored),
            None => crate::arrays::store_row(cx, row, zero, stored),
        }
    })
}

/// Stores stream bits `[start, end)` as the queue, dynamic array or string
/// `operand` (a `with` operand front-pads `pad` zero bits and takes `bits`:
/// the elements below the start are unspecified by the LRM, zero here). The
/// stream buffer is kept for the operands still to come.
fn consume_range<'ctx, 'd>(
    cx: &LowerCtx<'ctx, 'd, '_>,
    start: IntValue<'ctx>,
    end: IntValue<'ctx>,
    pad: Option<(IntValue<'ctx>, IntValue<'ctx>)>,
    operand: sv_lang::Expression<'d>,
    sink: Option<&mut CollSink<'ctx, 'd, '_>>,
) -> Result<(), LoweringError> {
    let i64t = cx.ctx.i64_type();
    rt(cx, CollOp::StreamEnter, 0, 1, 0, None);
    raw_call(
        cx,
        [0; 5],
        CollOp::StreamRange,
        i64t.const_zero(),
        Args {
            index: Some(start),
            index2: Some(end),
            ..Args::default()
        },
    );
    if let Some((pad, bits)) = pad {
        // Front-pad `pad` zero bits (negative start) for a `with` start != 0.
        raw_call(
            cx,
            [0; 5],
            CollOp::StreamRange,
            i64t.const_zero(),
            Args {
                index: Some(cx.b.build_int_neg(pad, "with_padneg").unwrap()),
                index2: Some(bits),
                ..Args::default()
            },
        );
    }
    store_stream_buffer(cx, operand, sink)?;
    rt(cx, CollOp::StreamLeave, 0, 0, 1, None);
    Ok(())
}

/// The run-time bit width (and front padding in bits) of a `with [a +: n]` /
/// `[a -: n]` / `[a : b]` stream target operand (a queue or dynamic array; LRM
/// 1800-2023 11.4.14.4): `n` elements are unpacked into elements `a ..
/// a + n - 1` and the array is resized to `a + n`; the elements below `a` are
/// unspecified by the LRM and become zero.
fn with_width<'ctx, 'd>(
    cx: &LowerCtx<'ctx, 'd, '_>,
    s: &sv_lang::StreamExpr<'d>,
) -> Result<
    (
        inkwell::values::IntValue<'ctx>,
        inkwell::values::IntValue<'ctx>,
    ),
    LoweringError,
> {
    use crate::expr::{RangeSelectionKindClass as K, classify_range_selection_kind};
    let unsup = |what: &str| LoweringError::unsupported_at(what, &cx.fn_sym);
    let operand = s.operand();
    let w = s.with_expr().expect("caller checked");
    if runtime_fixed_with(s) {
        // A fixed-size array: the elements stream from the range's start, no
        // padding.
        return Ok((
            crate::stream_with::runtime_bits(cx, operand, w)?,
            cx.ctx.i64_type().const_zero(),
        ));
    }
    let c = crate::collections::coll_of_expr(&operand)
        .filter(|c| !c.assoc && c.elem.is_integral())
        .ok_or_else(|| {
            unsup("a `with [...]` stream target that is not an integral queue or dynamic array")
        })?;
    let i64t = cx.ctx.i64_type();
    let int = |e: sv_lang::Expression<'d>| -> Result<_, LoweringError> {
        let pv = lower_expr(cx, e)?.into_packed("a stream `with` bound", &cx.fn_sym)?;
        let v = op(cx).lane_val(pv);
        Ok(if pv.width >= 64 {
            cx.b.build_int_truncate_or_bit_cast(v, i64t, "with_n")
                .unwrap()
        } else if pv.signed {
            cx.b.build_int_s_extend(v, i64t, "with_n").unwrap()
        } else {
            cx.b.build_int_z_extend(v, i64t, "with_n").unwrap()
        })
    };
    let (Some(l), Some(r), Some(kind)) =
        (w.range_left(), w.range_right(), w.range_selection_kind())
    else {
        return Err(unsup("a stream `with` that is not a range"));
    };
    let one = i64t.const_int(1, false);
    let zero = i64t.const_zero();
    let clamp0 = |v: inkwell::values::IntValue<'ctx>| {
        let neg =
            cx.b.build_int_compare(inkwell::IntPredicate::SLT, v, zero, "with_neg")
                .unwrap();
        cx.b.build_select(neg, zero, v, "with_c0")
            .unwrap()
            .into_int_value()
    };
    let lv = int(l)?;
    let rv = int(r)?;
    let (lo, count) = match classify_range_selection_kind(kind).map_err(|m| unsup(&m))? {
        K::IndexedUp => (lv, rv),
        K::IndexedDown => {
            // [a -: n] covers a-n+1 .. a.
            let lo = cx.b.build_int_sub(lv, rv, "with_lo").unwrap();
            (cx.b.build_int_add(lo, one, "with_lo1").unwrap(), rv)
        }
        K::Simple => {
            // [l : r] covers min(l, r) .. max(l, r).
            let gt =
                cx.b.build_int_compare(inkwell::IntPredicate::SGT, lv, rv, "with_gt")
                    .unwrap();
            let lo =
                cx.b.build_select(gt, rv, lv, "with_min")
                    .unwrap()
                    .into_int_value();
            let hi =
                cx.b.build_select(gt, lv, rv, "with_max")
                    .unwrap()
                    .into_int_value();
            let d = cx.b.build_int_sub(hi, lo, "with_d").unwrap();
            (lo, cx.b.build_int_add(d, one, "with_cnt").unwrap())
        }
    };
    let (lo, count) = (clamp0(lo), clamp0(count));
    let ew = i64t.const_int(u64::from(c.width), false);
    let bits = cx.b.build_int_mul(count, ew, "with_bits").unwrap();
    let pad = cx.b.build_int_mul(lo, ew, "with_pad").unwrap();
    Ok((bits, pad))
}