structured-zstd 0.0.54

// `format!` is used by ungated tests (e.g. the btlazy2 dict-reuse
// byte-identity test), so the import must not be feature-gated — under
// default features (no `dict-builder`) the gated form left `format!`
// unresolved when the test module is compiled.
use alloc::format;
use alloc::vec;

use super::FrameCompressor;
use crate::common::{MAGIC_NUM, MAX_BLOCK_SIZE};
use crate::decoding::FrameDecoder;
use crate::encoding::{Matcher, Sequence};
use alloc::vec::Vec;

fn generate_data(seed: u64, len: usize) -> Vec<u8> {
    let mut state = seed;
    let mut data = Vec::with_capacity(len);
    for _ in 0..len {
        state = state
            .wrapping_mul(6364136223846793005)
            .wrapping_add(1442695040888963407);
        data.push((state >> 33) as u8);
    }
    data
}

// Cross-implementation parity tests (compress here, decode through the C
// bindings) moved to `ffi-bench/tests/frame_compressor_ffi.rs` so the
// library crate never links libzstd.

struct NoDictionaryMatcher {
    last_space: Vec<u8>,
    window_size: u64,
}

impl NoDictionaryMatcher {
    fn new(window_size: u64) -> Self {
        Self {
            last_space: Vec::new(),
            window_size,
        }
    }
}

impl Matcher for NoDictionaryMatcher {
    fn get_next_space(&mut self) -> Vec<u8> {
        vec![0; self.window_size as usize]
    }

    fn get_last_space(&mut self) -> &[u8] {
        self.last_space.as_slice()
    }

    fn commit_space(&mut self, space: Vec<u8>) {
        self.last_space = space;
    }

    fn skip_matching(&mut self) {}

    fn start_matching(&mut self, mut handle_sequence: impl for<'a> FnMut(Sequence<'a>)) {
        handle_sequence(Sequence::Literals {
            literals: self.last_space.as_slice(),
        });
    }

    fn reset(&mut self, _level: super::CompressionLevel) {
        self.last_space.clear();
    }

    fn window_size(&self) -> u64 {
        self.window_size
    }
}

#[test]
fn frame_starts_with_magic_num() {
    let mock_data = [1_u8, 2, 3].as_slice();
    let mut output: Vec<u8> = Vec::new();
    let mut compressor = FrameCompressor::new(super::CompressionLevel::Uncompressed);
    compressor.set_source(mock_data);
    compressor.set_drain(&mut output);

    compressor.compress();
    assert!(output.starts_with(&MAGIC_NUM.to_le_bytes()));
}

#[test]
fn very_simple_raw_compress() {
    let mock_data = [1_u8, 2, 3].as_slice();
    let mut output: Vec<u8> = Vec::new();
    let mut compressor = FrameCompressor::new(super::CompressionLevel::Uncompressed);
    compressor.set_source(mock_data);
    compressor.set_drain(&mut output);

    compressor.compress();
}

#[test]
fn very_simple_compress() {
    let mut mock_data = vec![0; 1 << 17];
    mock_data.extend(vec![1; (1 << 17) - 1]);
    mock_data.extend(vec![2; (1 << 18) - 1]);
    mock_data.extend(vec![2; 1 << 17]);
    mock_data.extend(vec![3; (1 << 17) - 1]);
    let mut output: Vec<u8> = Vec::new();
    let mut compressor = FrameCompressor::new(super::CompressionLevel::Uncompressed);
    compressor.set_source(mock_data.as_slice());
    compressor.set_drain(&mut output);

    compressor.compress();

    let mut decoder = FrameDecoder::new();
    let mut decoded = Vec::with_capacity(mock_data.len());
    decoder.decode_all_to_vec(&output, &mut decoded).unwrap();
    assert_eq!(mock_data, decoded);
}

#[test]
fn rle_compress() {
    let mock_data = vec![0; 1 << 19];
    let mut output: Vec<u8> = Vec::new();
    let mut compressor = FrameCompressor::new(super::CompressionLevel::Uncompressed);
    compressor.set_source(mock_data.as_slice());
    compressor.set_drain(&mut output);

    compressor.compress();

    let mut decoder = FrameDecoder::new();
    let mut decoded = Vec::with_capacity(mock_data.len());
    decoder.decode_all_to_vec(&output, &mut decoded).unwrap();
    assert_eq!(mock_data, decoded);
}

#[test]
fn aaa_compress() {
    let mock_data = vec![0, 1, 3, 4, 5];
    let mut output: Vec<u8> = Vec::new();
    let mut compressor = FrameCompressor::new(super::CompressionLevel::Uncompressed);
    compressor.set_source(mock_data.as_slice());
    compressor.set_drain(&mut output);

    compressor.compress();

    let mut decoder = FrameDecoder::new();
    let mut decoded = Vec::with_capacity(mock_data.len());
    decoder.decode_all_to_vec(&output, &mut decoded).unwrap();
    assert_eq!(mock_data, decoded);
}

#[test]
fn dictionary_compression_sets_required_dict_id_and_roundtrips() {
    let dict_raw = include_bytes!("../../../dict_tests/dictionary");
    let dict_for_encoder = crate::decoding::Dictionary::decode_dict(dict_raw).unwrap();
    let dict_for_decoder = crate::decoding::Dictionary::decode_dict(dict_raw).unwrap();

    let mut data = Vec::new();
    for _ in 0..8 {
        data.extend_from_slice(&dict_for_decoder.dict_content[..2048]);
    }

    let mut with_dict = Vec::new();
    let mut compressor = FrameCompressor::new(super::CompressionLevel::Fastest);
    let previous = compressor
        .set_dictionary_from_bytes(dict_raw)
        .expect("dictionary bytes should parse");
    assert!(
        previous.is_none(),
        "first dictionary insert should return None"
    );
    assert_eq!(
        compressor
            .set_dictionary(dict_for_encoder)
            .expect("valid dictionary should attach")
            .expect("set_dictionary_from_bytes inserted previous dictionary")
            .id(),
        dict_for_decoder.id
    );
    compressor.set_source(data.as_slice());
    compressor.set_drain(&mut with_dict);
    compressor.compress();

    let (frame_header, _) = crate::decoding::frame::read_frame_header(with_dict.as_slice())
        .expect("encoded stream should have a frame header");
    assert_eq!(frame_header.dictionary_id(), Some(dict_for_decoder.id));

    let mut decoder = FrameDecoder::new();
    let mut missing_dict_target = Vec::with_capacity(data.len());
    let err = decoder
        .decode_all_to_vec(&with_dict, &mut missing_dict_target)
        .unwrap_err();
    assert!(
        matches!(
            &err,
            crate::decoding::errors::FrameDecoderError::DictNotProvided { .. }
        ),
        "dict-compressed stream should require dictionary id, got: {err:?}"
    );

    let mut decoder = FrameDecoder::new();
    decoder.add_dict(dict_for_decoder).unwrap();
    let mut decoded = Vec::with_capacity(data.len());
    decoder.decode_all_to_vec(&with_dict, &mut decoded).unwrap();
    assert_eq!(decoded, data);
}

#[cfg(all(feature = "dict-builder", feature = "std"))]
#[test]
fn dictionary_compression_roundtrips_with_dict_builder_dictionary() {
    use std::io::Cursor;

    let mut training = Vec::new();
    for idx in 0..256u32 {
        training.extend_from_slice(
            format!("tenant=demo table=orders key={idx} region=eu\n").as_bytes(),
        );
    }
    let mut raw_dict = Vec::new();
    crate::dictionary::create_raw_dict_from_source(
        Cursor::new(training.as_slice()),
        training.len(),
        &mut raw_dict,
        4096,
    )
    .expect("dict-builder training should succeed");
    assert!(
        !raw_dict.is_empty(),
        "dict-builder produced an empty dictionary"
    );

    let dict_id = 0xD1C7_0008;
    let encoder_dict =
        crate::decoding::Dictionary::from_raw_content(dict_id, raw_dict.clone()).unwrap();
    let decoder_dict =
        crate::decoding::Dictionary::from_raw_content(dict_id, raw_dict.clone()).unwrap();

    // Payload the trained dictionary genuinely covers: training lines in a
    // permuted order, so whole lines are dictionary matches while the
    // payload does not simply repeat its previous line. (A payload of
    // unseen `key=` values is NOT such a case: every line then matches the
    // previous one through the repeat offset, and upstream too compresses
    // it smaller WITHOUT the dictionary, the dictionary id and the
    // dictionary-tier parameters costing more than the first line's
    // literals.)
    let mut payload = Vec::new();
    for i in 0..96u32 {
        let idx = (i * 37 + 11) % 256;
        payload.extend_from_slice(
            format!("tenant=demo table=orders key={idx} region=eu\n").as_bytes(),
        );
    }

    let mut without_dict = Vec::new();
    let mut baseline = FrameCompressor::new(super::CompressionLevel::Fastest);
    baseline.set_source(payload.as_slice());
    baseline.set_drain(&mut without_dict);
    baseline.compress();

    let mut with_dict = Vec::new();
    let mut compressor = FrameCompressor::new(super::CompressionLevel::Fastest);
    compressor
        .set_dictionary(encoder_dict)
        .expect("valid dict-builder dictionary should attach");
    compressor.set_source(payload.as_slice());
    compressor.set_drain(&mut with_dict);
    compressor.compress();

    let (frame_header, _) = crate::decoding::frame::read_frame_header(with_dict.as_slice())
        .expect("encoded stream should have a frame header");
    assert_eq!(frame_header.dictionary_id(), Some(dict_id));
    let mut decoder = FrameDecoder::new();
    decoder.add_dict(decoder_dict).unwrap();
    let mut decoded = Vec::with_capacity(payload.len());
    decoder.decode_all_to_vec(&with_dict, &mut decoded).unwrap();
    assert_eq!(decoded, payload);
    assert!(
        with_dict.len() < without_dict.len(),
        "trained dictionary should improve compression for this small payload (with_dict={}, without_dict={})",
        with_dict.len(),
        without_dict.len(),
    );
}

#[test]
fn set_dictionary_from_bytes_seeds_entropy_tables_for_first_block() {
    let dict_raw = include_bytes!("../../../dict_tests/dictionary");
    let mut output = Vec::new();
    let input = b"";

    let mut compressor = FrameCompressor::new(super::CompressionLevel::Fastest);
    let previous = compressor
        .set_dictionary_from_bytes(dict_raw)
        .expect("dictionary bytes should parse");
    assert!(previous.is_none());

    compressor.set_source(input.as_slice());
    compressor.set_drain(&mut output);
    compressor.compress();

    assert!(
        compressor.state.last_huff_table.is_some(),
        "dictionary entropy should seed previous huffman table before first block"
    );
    assert!(
        compressor.state.fse_tables.ll_previous.is_some(),
        "dictionary entropy should seed previous ll table before first block"
    );
    assert!(
        compressor.state.fse_tables.ml_previous.is_some(),
        "dictionary entropy should seed previous ml table before first block"
    );
    assert!(
        compressor.state.fse_tables.of_previous.is_some(),
        "dictionary entropy should seed previous of table before first block"
    );
}

// `set_content_size_flag(false)`: the header must omit the FCS field
// (and the single-segment layout that requires it) while the frame
// still round-trips through our decoder.
#[test]
fn content_size_flag_off_omits_fcs_and_roundtrips() {
    let payload = alloc::vec![0x42u8; 4096];

    let mut compressor: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Fastest);
    let mut with_fcs = Vec::new();
    compressor.compress_independent_frame_into(&payload, &mut with_fcs);

    compressor.set_content_size_flag(false);
    let mut without_fcs = Vec::new();
    compressor.compress_independent_frame_into(&payload, &mut without_fcs);

    let parsed_with = crate::decoding::frame::read_frame_header(with_fcs.as_slice())
        .expect("flag-on frame header must parse")
        .0;
    assert_eq!(parsed_with.frame_content_size(), 4096);

    let parsed_without = crate::decoding::frame::read_frame_header(without_fcs.as_slice())
        .expect("flag-off frame header must parse")
        .0;
    // 0 is the decoder's "unknown content size" sentinel...
    assert_eq!(
        parsed_without.frame_content_size(),
        0,
        "FCS must be omitted with the content-size flag off"
    );
    // ...and the descriptor must confirm the field is ABSENT (0 bytes),
    // not present with an explicit zero value.
    assert_eq!(
        parsed_without
            .descriptor
            .frame_content_size_bytes()
            .expect("descriptor must parse"),
        0,
        "the FCS field itself must be omitted, not written as zero"
    );

    let mut decoder = crate::decoding::FrameDecoder::new();
    // `decode_all_to_vec` fills existing capacity (no FCS to pre-size
    // from with the flag off), so reserve the expected payload upfront.
    let mut decoded = Vec::with_capacity(payload.len() + 64);
    decoder
        .decode_all_to_vec(&without_fcs, &mut decoded)
        .expect("flag-off frame must decode");
    assert_eq!(decoded, payload);
}

// `set_dictionary_id_flag(false)`: a dict-compressed frame must omit
// the dictionary ID and still decode when the dictionary is handed to
// the decoder explicitly.
#[test]
fn dict_id_flag_off_omits_dictionary_id_and_roundtrips() {
    let dict_raw = include_bytes!("../../../dict_tests/dictionary");
    let payload = b"dictionary-keyed payload dictionary-keyed payload".repeat(8);

    let mut compressor: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Fastest);
    compressor
        .set_dictionary_from_bytes(dict_raw)
        .expect("dictionary bytes should parse");
    compressor.set_dictionary_id_flag(false);
    let mut frame = Vec::new();
    compressor.compress_independent_frame_into(&payload, &mut frame);

    let parsed = crate::decoding::frame::read_frame_header(frame.as_slice())
        .expect("frame header must parse")
        .0;
    assert_eq!(
        parsed.dictionary_id(),
        None,
        "dictionary id must be omitted with the dict-id flag off"
    );

    // With the ID omitted the decoder cannot look the dictionary up by
    // header; hand it explicitly (the `reset_with_dict_handle` path).
    let mut sd =
        crate::decoding::StreamingDecoder::new_with_dictionary_bytes(frame.as_slice(), dict_raw)
            .expect("decoder must accept the dictionary");
    let mut dec = Vec::new();
    std::io::Read::read_to_end(&mut sd, &mut dec)
        .expect("frame must decode with the dictionary handed explicitly");
    assert_eq!(dec, payload);
}

// The output reservation must track the observed compression ratio, not
// the whole-input `compress_bound`: a multi-MiB compressible stream's
// output buffer stays at output scale (the old up-front bound held an
// input-sized allocation for the whole frame). Incompressible input may
// still re-estimate to ~the full bound — that is the genuine worst case.
#[test]
fn compressible_stream_output_capacity_stays_at_output_scale() {
    // 4 MiB of highly repetitive log-like lines.
    let line = b"ts=2026-03-26T21:39:28Z level=INFO msg=\"flush memtable\" tenant=demo\n";
    let mut input = Vec::with_capacity(4 << 20);
    while input.len() < (4 << 20) {
        let take = line.len().min((4 << 20) - input.len());
        input.extend_from_slice(&line[..take]);
    }

    let mut compressor: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Fastest);
    let mut out = Vec::new();
    compressor.compress_independent_frame_into(&input, &mut out);

    assert!(!out.is_empty());
    assert!(
        out.capacity() < input.len() / 4,
        "capacity {} must stay at output scale (input {}, output {})",
        out.capacity(),
        input.len(),
        out.len()
    );

    // Round-trip: the adaptive reservation must not affect the bytes.
    let mut decoder = crate::decoding::FrameDecoder::new();
    let mut decoded = Vec::with_capacity(input.len() + 64);
    decoder
        .decode_all_to_vec(&out, &mut decoded)
        .expect("frame must decode");
    assert_eq!(decoded, input);
}

// A dictionary frame with a known content size that fits the window
// must take the single-segment layout (reference parity): the
// dictionary is decoder setup state, not part of the regenerated
// segment, so it must not force the windowed multi-segment layout.
#[test]
fn dict_frame_with_known_size_is_single_segment() {
    let dict_raw = include_bytes!("../../../dict_tests/dictionary");
    let payload = b"dictionary-keyed payload dictionary-keyed payload".repeat(64);

    let mut compressor: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Fastest);
    compressor
        .set_dictionary_from_bytes(dict_raw)
        .expect("dictionary bytes should parse");
    let mut frame = Vec::new();
    compressor.compress_independent_frame_into(&payload, &mut frame);

    let parsed = crate::decoding::frame::read_frame_header(frame.as_slice())
        .expect("frame header must parse")
        .0;
    assert!(
        parsed.descriptor.single_segment_flag(),
        "dict frame with known size <= window must be single-segment"
    );
    assert!(parsed.dictionary_id().is_some());
    assert_eq!(parsed.frame_content_size(), payload.len() as u64);

    // Round-trip through our own decoder with the dictionary.
    let mut decoder = crate::decoding::FrameDecoder::new();
    decoder
        .add_dict_from_bytes(dict_raw)
        .expect("decoder must accept the dictionary");
    let mut decoded = Vec::with_capacity(payload.len() + 64);
    decoder
        .decode_all_to_vec(&frame, &mut decoded)
        .expect("single-segment dict frame must decode");
    assert_eq!(decoded, payload);
}

// Regression: after `clear_dictionary()` a reused compressor must fully
// deactivate the dictionary state. The dict-active matcher reset rewinds the
// hash/chain tables to the origin (`position_base = 0`) and DEFERS the table
// clear to the next prime/restore. If `clear_dictionary` leaves the matcher
// marked dictionary-active, a subsequent NO-dictionary frame hits that
// deferred-clear branch but never runs prime/restore, so stale dict-region
// entries (old absolute positions) survive at the rewound base and can
// surface as bogus matches. The no-dict frame must still round-trip without
// the dictionary. Uses Level(16) (btopt → HashChain backend, whose storage
// owns the deferred-clear path).
#[test]
fn clear_dictionary_then_nodict_frame_roundtrips() {
    let dict_raw = include_bytes!("../../../dict_tests/dictionary");
    // Payload B embeds dictionary bytes up front so a surviving dict-region
    // chain entry would be a tempting (wrong) match candidate.
    let mut payload_b = Vec::new();
    payload_b.extend_from_slice(&dict_raw[..dict_raw.len().min(2048)]);
    payload_b.extend_from_slice(
        b"no-dictionary tail no-dictionary tail"
            .repeat(16)
            .as_slice(),
    );

    let mut compressor: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Level(16));
    compressor
        .set_dictionary_from_bytes(dict_raw)
        .expect("dictionary bytes should parse");
    // Frame A primes the dictionary (sets the matcher dictionary-active).
    let mut frame_a = Vec::new();
    compressor.compress_independent_frame_into(
        b"dictionary-keyed payload dictionary-keyed payload"
            .repeat(8)
            .as_slice(),
        &mut frame_a,
    );
    // Remove the dictionary, then compress a no-dictionary frame on the
    // SAME reused compressor.
    compressor.clear_dictionary();
    let mut frame_b = Vec::new();
    compressor.compress_independent_frame_into(&payload_b, &mut frame_b);

    // Frame B must decode WITHOUT any dictionary.
    let mut decoder = crate::decoding::FrameDecoder::new();
    let mut decoded = Vec::with_capacity(payload_b.len() + 64);
    decoder
        .decode_all_to_vec(&frame_b, &mut decoded)
        .expect("no-dict frame after clear_dictionary must decode");
    assert_eq!(
        decoded, payload_b,
        "no-dict frame after clear_dictionary must round-trip exactly"
    );
}

/// The weight builder keeps its buffers between blocks and frames — that is
/// what it is for — so a compressor asked how much it holds has to count them.
/// Reported through the C API as `ZSTD_sizeof_CCtx`, a caller budgeting memory
/// around a reused context would otherwise be told less than it keeps.
#[test]
fn heap_size_counts_the_retained_weight_scratch() {
    // Literal-heavy with a skewed alphabet: the matcher finds little to
    // repeat, so the block goes out as literals and the entropy path actually
    // builds a Huffman table — which is what leaves the scratch holding its
    // buffers. A periodic fixture would compress to sequences instead and
    // never reach the table build.
    let mut state = 0x2545_F491_4F6C_DD1Du64;
    let data: Vec<u8> = (0..64 * 1024u32)
        .map(|_| {
            state = state.wrapping_mul(6364136223846793005).wrapping_add(1);
            ((state >> 33) % 64) as u8
        })
        .collect();
    let mut compressor: FrameCompressor<&[u8]> =
        FrameCompressor::new(super::CompressionLevel::Level(3));
    let before = compressor.heap_size();
    compressor.set_source(data.as_slice());
    compressor.set_drain(Vec::new());
    compressor.compress();

    let scratch_bytes = compressor.state.huff_weights.heap_size();
    assert!(
        scratch_bytes > 0,
        "a frame that built a table leaves the weight buffers allocated"
    );
    assert!(
        compressor.heap_size() > before,
        "compressing grows the reported footprint"
    );

    // Isolate the scratch's contribution: take it away and the reported total
    // must fall by exactly what it held. Comparing against the pre-frame total
    // would not — the matcher's tables grew over the same frame and would hide
    // an unreported scratch behind them.
    let with_scratch = compressor.heap_size();
    let taken = core::mem::take(&mut compressor.state.huff_weights);
    let without_scratch = compressor.heap_size();
    assert_eq!(
        with_scratch - without_scratch,
        taken.heap_size(),
        "the reported footprint has to move by exactly what the scratch holds"
    );
}

// Regression test: `heap_size()` must count the retained Huffman tables
// (the active `last_huff_table` and the recycled `huff_table_spare`).
// A reused context that parks a table would otherwise under-report its
// footprint through the public size API.
#[test]
fn heap_size_counts_active_and_spare_huffman_tables() {
    let mut compressor: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Fastest);
    let base = compressor.heap_size();

    let active = crate::huff0::huff0_encoder::HuffmanTable::build_from_data(
        b"abacabadabacabaeabacabadabacaba",
    );
    let active_bytes = active.heap_size();
    assert!(active_bytes > 0, "built table must own heap buffers");
    compressor.state.last_huff_table = Some(active);
    assert_eq!(
        compressor.heap_size(),
        base + active_bytes,
        "heap_size must include the active last_huff_table"
    );

    let spare = crate::huff0::huff0_encoder::HuffmanTable::build_from_data(
        b"the quick brown fox jumps over the lazy dog",
    );
    let spare_bytes = spare.heap_size();
    assert!(spare_bytes > 0, "built table must own heap buffers");
    compressor.state.huff_table_spare = Some(spare);
    assert_eq!(
        compressor.heap_size(),
        base + active_bytes + spare_bytes,
        "heap_size must include the parked huff_table_spare"
    );
}

#[test]
fn set_encoder_dictionary_reattaches_prepared_dict_without_reparse() {
    let dict_raw = include_bytes!("../../../dict_tests/dictionary");
    let payload = b"tenant=demo table=orders op=put key=1 value=aaaaabbbbbcccccdddddeeeee\n\
              tenant=demo table=orders op=put key=2 value=aaaaabbbbbcccccdddddeeeee\n";

    // Prepare the EncoderDictionary once, then attach it via the prepared-
    // dictionary API (no raw-blob reparse at attach time).
    let prepared = super::EncoderDictionary::from_bytes(dict_raw).expect("dict bytes should parse");
    let dict_id = prepared.id();

    let mut with_dict = Vec::new();
    let mut compressor = FrameCompressor::new(super::CompressionLevel::Fastest);
    let previous = compressor
        .set_encoder_dictionary(prepared)
        .expect("prepared dictionary should attach");
    assert!(previous.is_none());
    compressor.set_source(payload.as_slice());
    compressor.set_drain(&mut with_dict);
    compressor.compress();
    // clear_dictionary hands the prepared dictionary back (last use of
    // `compressor`, so its `&mut with_dict` drain borrow ends here).
    let returned = compressor
        .clear_dictionary()
        .expect("dictionary was attached");
    assert_eq!(returned.id(), dict_id);

    // The reattached dictionary drives the frame: its id is advertised and
    // the stream round-trips through a decoder primed with the same dict.
    let (frame_header, _) = crate::decoding::frame::read_frame_header(with_dict.as_slice())
        .expect("encoded stream should have a frame header");
    assert_eq!(frame_header.dictionary_id(), Some(dict_id));
    let decoder_dict = crate::decoding::Dictionary::decode_dict(dict_raw).unwrap();
    let mut decoder = FrameDecoder::new();
    decoder.add_dict(decoder_dict).unwrap();
    let mut decoded = Vec::with_capacity(payload.len());
    decoder.decode_all_to_vec(&with_dict, &mut decoded).unwrap();
    assert_eq!(decoded.as_slice(), payload.as_slice());

    // The dictionary handed back by clear_dictionary reattaches to another
    // compressor without touching the raw bytes again, producing an
    // identical frame.
    let mut with_dict2 = Vec::new();
    let mut compressor2 = FrameCompressor::new(super::CompressionLevel::Fastest);
    compressor2
        .set_encoder_dictionary(returned)
        .expect("returned dictionary should reattach");
    compressor2.set_source(payload.as_slice());
    compressor2.set_drain(&mut with_dict2);
    compressor2.compress();
    assert_eq!(
        with_dict2, with_dict,
        "reattached prepared dict must produce an identical frame"
    );
}

#[test]
fn dict_primed_matcher_snapshot_reused_across_frames_is_byte_identical() {
    // CDict-equivalent: a compressor reused across frames with the same
    // dictionary restores the primed matcher snapshot on frames 2..N
    // (a table copy) instead of re-hashing the dictionary. The restored
    // state must reproduce the first-frame (freshly-primed) output
    // byte-for-byte, and every frame must round-trip through a
    // dict-primed decoder.
    let dict_raw = include_bytes!("../../../dict_tests/dictionary");
    // Source must exceed the Fast strategy's 8 KiB attach cutoff so the
    // copy-snapshot (restore) path is taken on frame 2 — at or below the
    // cutoff the upstream zstd attaches by reference and we fall back to re-prime,
    // which would not exercise restore.
    let mut payload = Vec::new();
    while payload.len() < 16 * 1024 {
        payload.extend_from_slice(
            b"tenant=demo table=orders op=put key=1 value=aaaaabbbbbcccccdddddeeeee\n",
        );
    }

    let prepared = super::EncoderDictionary::from_bytes(dict_raw).expect("dict bytes should parse");
    let dict_id = prepared.id();
    let mut compressor: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Fastest);
    compressor
        .set_encoder_dictionary(prepared)
        .expect("prepared dictionary should attach");

    // Frame 1 primes + captures the snapshot; frame 2 restores it.
    let frame1 = compressor.compress_independent_frame(payload.as_slice());
    let frame2 = compressor.compress_independent_frame(payload.as_slice());
    assert_eq!(
        frame1, frame2,
        "restored prime snapshot must reproduce the freshly-primed frame byte-for-byte"
    );

    // Both frames advertise the dict id and round-trip through a
    // dict-primed decoder.
    for frame in [&frame1, &frame2] {
        let (hdr, _) =
            crate::decoding::frame::read_frame_header(frame.as_slice()).expect("frame header");
        assert_eq!(hdr.dictionary_id(), Some(dict_id));
        let mut decoder = FrameDecoder::new();
        decoder
            .add_dict(crate::decoding::Dictionary::decode_dict(dict_raw).unwrap())
            .unwrap();
        let mut decoded = Vec::with_capacity(payload.len());
        decoder.decode_all_to_vec(frame, &mut decoded).unwrap();
        assert_eq!(decoded.as_slice(), payload.as_slice());
    }
}

#[test]
fn dict_primed_matcher_cache_reused_across_small_attach_frames_is_byte_identical() {
    // CDict-equivalent ATTACH path (small source, at/below the Fast 8 KiB
    // attach cutoff): frames 2..N re-prime — re-committing the dict bytes
    // to history — but reuse the already-built dict table instead of
    // re-hashing it. The cached-table frame must reproduce the
    // freshly-primed first frame byte-for-byte, and a fresh single-frame
    // compressor (no prior dict cache) must produce the identical bytes
    // too, proving the cache changes timing, not output.
    let dict_raw = include_bytes!("../../../dict_tests/dictionary");
    // Stay under the 8 KiB cutoff so the attach (re-prime) path is taken
    // every frame rather than the copy-snapshot restore.
    let mut payload = Vec::new();
    while payload.len() < 2 * 1024 {
        payload.extend_from_slice(b"tenant=demo op=put key=1 value=aaaaabbbbbcccccddddd\n");
    }

    let prepared = super::EncoderDictionary::from_bytes(dict_raw).expect("dict bytes should parse");
    let dict_id = prepared.id();
    let mut compressor: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Fastest);
    compressor
        .set_encoder_dictionary(prepared)
        .expect("prepared dictionary should attach");

    // Frame 1 builds + marks the dict table; frame 2 reuses it.
    let frame1 = compressor.compress_independent_frame(payload.as_slice());
    let frame2 = compressor.compress_independent_frame(payload.as_slice());
    assert_eq!(
        frame1, frame2,
        "reused dict table (attach path) must reproduce the freshly-built frame byte-for-byte"
    );

    // A fresh compressor (cold dict cache) must emit the same bytes — the
    // cache is a timing optimization, never a content change.
    let fresh_prepared =
        super::EncoderDictionary::from_bytes(dict_raw).expect("dict bytes should parse");
    let mut fresh: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Fastest);
    fresh
        .set_encoder_dictionary(fresh_prepared)
        .expect("prepared dictionary should attach");
    let fresh_frame = fresh.compress_independent_frame(payload.as_slice());
    assert_eq!(
        fresh_frame, frame1,
        "cold-cache compressor must match the warm-cache frame byte-for-byte"
    );

    for frame in [&frame1, &frame2] {
        let (hdr, _) =
            crate::decoding::frame::read_frame_header(frame.as_slice()).expect("frame header");
        assert_eq!(hdr.dictionary_id(), Some(dict_id));
        let mut decoder = FrameDecoder::new();
        decoder
            .add_dict(crate::decoding::Dictionary::decode_dict(dict_raw).unwrap())
            .unwrap();
        let mut decoded = Vec::with_capacity(payload.len());
        decoder.decode_all_to_vec(frame, &mut decoded).unwrap();
        assert_eq!(decoded.as_slice(), payload.as_slice());
    }
}

#[test]
fn dict_reused_across_many_lazy_frames_stays_applied() {
    // Regression: a reused HashChain-backed dictionary frame (lazy levels)
    // re-primes the dict at the live `history_abs_start` every frame. The
    // floor-advance reset let that base climb frame-over-frame until the
    // freshly-primed dict region dropped below `window_low`, after which
    // every dict match was silently lost and the output ballooned to the
    // no-dict size (observed at frame 3-4 of a reused compressor). Drive
    // many frames and require every one to stay byte-identical to the
    // first — the dict must keep applying, not decay after a few frames.
    // Multiple distinct lines so the dictionary is load-bearing: without it
    // frame 0 must emit each distinct line as literals; with it those lines
    // match the primed dict immediately. A single repeated line matches
    // in-frame regardless and would hide the regression.
    let lines: &[&[u8]] = &[
        b"ts=2026 level=INFO msg=\"flush memtable\" tenant=demo table=orders\n",
        b"ts=2026 level=INFO msg=\"rotate segment\" tenant=demo table=orders\n",
        b"ts=2026 level=INFO msg=\"compact level\" tenant=demo table=orders\n",
        b"ts=2026 level=INFO msg=\"write block\" tenant=demo table=orders\n",
    ];
    let fill = |n: usize| -> Vec<u8> {
        let mut b = Vec::with_capacity(n);
        while b.len() < n {
            for l in lines {
                if b.len() >= n {
                    break;
                }
                let take = (n - b.len()).min(l.len());
                b.extend_from_slice(&l[..take]);
            }
        }
        b
    };
    let dict = fill(8 * 1024);
    let payload = fill(16 * 1024);
    let dict_obj =
        crate::decoding::Dictionary::from_raw_content(1, dict).expect("raw dict should build");

    let mut compressor: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Level(6));
    compressor.set_dictionary_id_flag(false);
    compressor
        .set_dictionary(dict_obj)
        .expect("dict should attach");

    let first = compressor.compress_independent_frame(payload.as_slice());

    // No-dict baseline at the same level: the dictionary must be
    // load-bearing, so the dict-applied frame has to beat it. Without this
    // anchor the equal-length loop below would also pass if EVERY frame
    // decayed to the no-dict size in lockstep.
    let mut nodict: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Level(6));
    let no_dict_frame = nodict.compress_independent_frame(payload.as_slice());
    assert!(
        first.len() < no_dict_frame.len(),
        "dict must be load-bearing: dict frame {} should beat the no-dict baseline {}",
        first.len(),
        no_dict_frame.len(),
    );

    for i in 1..16 {
        let frame = compressor.compress_independent_frame(payload.as_slice());
        // Byte-identity, not just equal length: a same-size divergence (e.g.
        // a different match decision once the resident dict bookkeeping
        // drifts) would slip past a length-only check.
        assert_eq!(
            frame, first,
            "frame {i} of a reused dict compressor must stay byte-identical to \
                 the first (dict still applied, no decay or bookkeeping drift)"
        );
    }
}

#[test]
fn dict_fast_epoch_reset_many_frames_and_attach_copy_alternation_byte_identical() {
    // The Fast attach path invalidates the main hash table between
    // frames with an epoch-bias advance instead of a memset. Two things
    // need proving against a fresh-compressor reference:
    // 1. the bias accumulates across MANY reused frames without ever
    //    letting a stale entry through (every frame byte-identical);
    // 2. crossing the 8 KiB attach/copy cutoff in both directions
    //    (attach → copy clears the bias for the raw-slice kernel,
    //    copy → attach re-enters epoch mode) stays byte-identical.
    let dict_raw = include_bytes!("../../../dict_tests/dictionary");
    let mut small = Vec::new();
    while small.len() < 2 * 1024 {
        small.extend_from_slice(b"tenant=demo op=put key=1 value=aaaaabbbbbcccccddddd\n");
    }
    // Over the Fast 8 KiB attach cutoff → copy-mode frame.
    let mut large = Vec::new();
    while large.len() < 64 * 1024 {
        large.extend_from_slice(b"tenant=demo op=scan range=[k0,k9) limit=500 order=asc\n");
    }

    let mut reused: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Fastest);
    reused
        .set_encoder_dictionary(
            super::EncoderDictionary::from_bytes(dict_raw).expect("dict bytes should parse"),
        )
        .expect("prepared dictionary should attach");

    let reference = |payload: &[u8]| -> alloc::vec::Vec<u8> {
        let mut fresh: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Fastest);
        fresh
            .set_encoder_dictionary(
                super::EncoderDictionary::from_bytes(dict_raw).expect("dict bytes should parse"),
            )
            .expect("prepared dictionary should attach");
        fresh.compress_independent_frame(payload)
    };

    let small_expected = reference(&small);
    let large_expected = reference(&large);

    // 1. Long attach-only run: every frame advances the epoch bias.
    for i in 0..32 {
        let frame = reused.compress_independent_frame(small.as_slice());
        assert_eq!(
            frame, small_expected,
            "attach frame {i} diverged from the fresh-compressor reference"
        );
    }
    // 2. Cutoff alternation: attach → copy → attach → copy.
    for i in 0..4 {
        let frame = reused.compress_independent_frame(large.as_slice());
        assert_eq!(
            frame, large_expected,
            "copy frame {i} diverged from the fresh-compressor reference"
        );
        let frame = reused.compress_independent_frame(small.as_slice());
        assert_eq!(
            frame, small_expected,
            "attach frame after copy {i} diverged from the fresh-compressor reference"
        );
    }
}

#[test]
fn dict_primed_btlazy2_reused_across_attach_and_copy_boundary_is_byte_identical() {
    // Btlazy2 (Level 15) uses the 32 KiB dict attach/copy cutoff in
    // prepare_frame. Exercise BOTH sides of that boundary on a reused
    // compressor: a sub-cutoff payload (re-prime/attach path) and an
    // over-cutoff payload (copy-snapshot restore path). In each case the
    // warm-cache second frame must reproduce the cold-cache first frame
    // byte-for-byte (the dict cache is a timing optimization, never a
    // content change), and every frame must round-trip.
    let dict_raw = include_bytes!("../../../dict_tests/dictionary");
    let dict_id = super::EncoderDictionary::from_bytes(dict_raw)
        .expect("dict bytes should parse")
        .id();
    // Distinct lines so the payload does not trivially self-compress; the
    // BT finder + dict dual-probe both get exercised.
    let make_payload = |target: usize| {
        let mut p = Vec::with_capacity(target);
        let mut i = 0u64;
        while p.len() < target {
            p.extend_from_slice(
                format!(
                    "tenant=demo op=put key={i} value=aaaaabbbbbcccccddddd-{}\n",
                    i % 97
                )
                .as_bytes(),
            );
            i += 1;
        }
        p
    };
    // Below the 32 KiB cutoff (attach/re-prime) and above it (copy-snapshot).
    for target in [16 * 1024usize, 64 * 1024usize] {
        let payload = make_payload(target);
        let mut warm: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Level(15));
        warm.set_encoder_dictionary(
            super::EncoderDictionary::from_bytes(dict_raw).expect("dict parse"),
        )
        .expect("dict attach");
        // Frame 1 builds + marks the dict tables; frame 2 reuses them.
        let frame1 = warm.compress_independent_frame(payload.as_slice());
        let frame2 = warm.compress_independent_frame(payload.as_slice());
        assert_eq!(
            frame1, frame2,
            "reused dict cache must reproduce the freshly-primed frame byte-for-byte \
                 (Level 15, target={target})"
        );
        // Cold-cache compressor: must match the warm-cache bytes.
        let mut cold: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Level(15));
        cold.set_encoder_dictionary(
            super::EncoderDictionary::from_bytes(dict_raw).expect("dict parse"),
        )
        .expect("dict attach");
        let cold_frame = cold.compress_independent_frame(payload.as_slice());
        assert_eq!(
            cold_frame, frame1,
            "cold-cache compressor must match warm-cache frame (Level 15, target={target})"
        );
        // Round-trip through a decoder primed with the same dict.
        for frame in [&frame1, &frame2] {
            let (hdr, _) =
                crate::decoding::frame::read_frame_header(frame.as_slice()).expect("frame header");
            assert_eq!(hdr.dictionary_id(), Some(dict_id));
            let mut decoder = FrameDecoder::new();
            decoder
                .add_dict(crate::decoding::Dictionary::decode_dict(dict_raw).unwrap())
                .unwrap();
            let mut decoded = Vec::with_capacity(payload.len());
            decoder.decode_all_to_vec(frame, &mut decoded).unwrap();
            assert_eq!(decoded.as_slice(), payload.as_slice());
        }
    }
}

#[test]
fn dict_primed_btultra2_restore_is_floor_safe_and_byte_identical() {
    // Regression guard for the dictionary primed-snapshot RESTORE path on
    // the binary-tree (btultra2 / Level 22) backend — the path a minimal /
    // decoupled prepared-dict refactor rewrites.
    //
    // The trap it pins: a reused compressor compresses frame A (which fills
    // the live hash/chain tables with frame-A positions and advances the
    // window floor), then frame B of the SAME resolved shape (same size →
    // same PrimedKey → the snapshot RESTORE path) but DIFFERENT content. The
    // restore must reinstate the clean post-prime dict state with NO live
    // frame-A entries surviving above the restored floor; a restore that
    // leaks stale frame-A positions would surface FALSE matches and produce
    // a different (or undecodable) frame B. The invariant: a snapshot
    // restore is a pure timing optimization and MUST be byte-identical to a
    // cold compressor compressing frame B from scratch, and must round-trip.
    let dict_raw = include_bytes!("../../../dict_tests/dictionary");
    let dict_id = super::EncoderDictionary::from_bytes(dict_raw)
        .expect("dict bytes should parse")
        .id();
    // 48 KiB > the btultra2 8 KiB attach cutoff → the copy-snapshot
    // capture/restore path. Two distinct payloads of the SAME size so frame
    // B resolves to frame A's snapshot key and takes the restore path.
    let make_payload = |seed: u64, target: usize| {
        let mut p = Vec::with_capacity(target);
        let mut i = seed;
        while p.len() < target {
            p.extend_from_slice(
                format!(
                    "tenant=demo op=put key={i} value=aaaaabbbbbcccccddddd-{}\n",
                    i % 89
                )
                .as_bytes(),
            );
            i = i.wrapping_add(1);
        }
        p.truncate(target);
        p
    };
    let size = 48 * 1024usize;
    let frame_a = make_payload(0, size);
    let frame_b = make_payload(1_000_000, size);

    let mut warm: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Level(22));
    warm.set_encoder_dictionary(
        super::EncoderDictionary::from_bytes(dict_raw).expect("dict parse"),
    )
    .expect("dict attach");
    // Frame A: cold cache — primes the dict + captures the snapshot, and
    // fills the live tables with frame-A positions.
    let _wa = warm.compress_independent_frame(frame_a.as_slice());
    // Frame B: warm cache — takes the snapshot RESTORE path (same size).
    let warm_b = warm.compress_independent_frame(frame_b.as_slice());

    // Cold compressor compressing frame B from scratch: the ground truth.
    let mut cold: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Level(22));
    cold.set_encoder_dictionary(
        super::EncoderDictionary::from_bytes(dict_raw).expect("dict parse"),
    )
    .expect("dict attach");
    let cold_b = cold.compress_independent_frame(frame_b.as_slice());

    assert_eq!(
        warm_b, cold_b,
        "frame B via snapshot restore must be byte-identical to a cold compress \
             (a restore that leaks frame-A live-table entries would diverge here)"
    );

    // Round-trip frame B through a dict-primed decoder.
    let (hdr, _) =
        crate::decoding::frame::read_frame_header(warm_b.as_slice()).expect("frame header");
    assert_eq!(hdr.dictionary_id(), Some(dict_id));
    let mut decoder = FrameDecoder::new();
    decoder
        .add_dict(crate::decoding::Dictionary::decode_dict(dict_raw).unwrap())
        .unwrap();
    let mut decoded = Vec::with_capacity(frame_b.len());
    decoder
        .decode_all_to_vec(warm_b.as_slice(), &mut decoded)
        .unwrap();
    assert_eq!(decoded.as_slice(), frame_b.as_slice());
}

#[test]
fn dict_primed_btultra2_ldm_restore_is_byte_identical() {
    // Same restore-path byte-identity guard as
    // `dict_primed_btultra2_restore_is_floor_safe_and_byte_identical`, but
    // with long-distance matching ENABLED. The BtMatcher's LDM producer is
    // part of the snapshot; a refactor that decouples it (so the snapshot
    // does not retain the empty LDM table) must reinstate an equivalent
    // empty producer on restore. This pins that the warm-cache (restore)
    // frame stays byte-identical to a cold compress when LDM is on.
    let dict_raw = include_bytes!("../../../dict_tests/dictionary");
    let dict_id = super::EncoderDictionary::from_bytes(dict_raw)
        .expect("dict bytes should parse")
        .id();
    let make_payload = |seed: u64, target: usize| {
        let mut p = Vec::with_capacity(target);
        let mut i = seed;
        while p.len() < target {
            p.extend_from_slice(
                format!(
                    "tenant=demo op=put key={i} value=aaaaabbbbbcccccddddd-{}\n",
                    i % 89
                )
                .as_bytes(),
            );
            i = i.wrapping_add(1);
        }
        p.truncate(target);
        p
    };
    let ldm_params =
        crate::encoding::CompressionParameters::builder(super::CompressionLevel::Level(22))
            .enable_long_distance_matching(true)
            .build()
            .expect("LDM-only params build");
    let size = 48 * 1024usize;
    let frame_a = make_payload(0, size);
    let frame_b = make_payload(1_000_000, size);

    let mut warm: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Level(22));
    warm.set_parameters(&ldm_params);
    warm.set_encoder_dictionary(
        super::EncoderDictionary::from_bytes(dict_raw).expect("dict parse"),
    )
    .expect("dict attach");
    let _wa = warm.compress_independent_frame(frame_a.as_slice());
    let warm_b = warm.compress_independent_frame(frame_b.as_slice());

    let mut cold: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Level(22));
    cold.set_parameters(&ldm_params);
    cold.set_encoder_dictionary(
        super::EncoderDictionary::from_bytes(dict_raw).expect("dict parse"),
    )
    .expect("dict attach");
    let cold_b = cold.compress_independent_frame(frame_b.as_slice());

    assert_eq!(
        warm_b, cold_b,
        "LDM-on frame B via snapshot restore must be byte-identical to a cold compress"
    );

    let (hdr, _) =
        crate::decoding::frame::read_frame_header(warm_b.as_slice()).expect("frame header");
    assert_eq!(hdr.dictionary_id(), Some(dict_id));
    let mut decoder = FrameDecoder::new();
    decoder
        .add_dict(crate::decoding::Dictionary::decode_dict(dict_raw).unwrap())
        .unwrap();
    let mut decoded = Vec::with_capacity(frame_b.len());
    decoder
        .decode_all_to_vec(warm_b.as_slice(), &mut decoded)
        .unwrap();
    assert_eq!(decoded.as_slice(), frame_b.as_slice());
}

#[test]
fn set_dictionary_from_bytes_matches_full_decode_byte_for_byte() {
    // The encoder-only dict parse (`decode_dict_for_encoding`, used by
    // `set_dictionary_from_bytes`) skips the FSE/HUF decoder-table build and
    // the enrich passes. The encoder entropy tables are derived purely from
    // the symbol probabilities / Huffman weights, so the compressed output
    // MUST be byte-identical to the full-decode path. This pins the
    // load-bearing equivalence so a future FSE/HUF parsing refactor that
    // still round-trips but silently diverges on the probabilities/weights
    // fails loudly here instead of producing a different (but valid) frame.
    let dict_raw = include_bytes!("../../../dict_tests/dictionary");
    let payload = b"tenant=demo table=orders op=put key=1 value=aaaaabbbbbcccccdddddeeeee\n\
              tenant=demo table=orders op=put key=2 value=aaaaabbbbbcccccdddddeeeee\n";

    // Path A: encoder-only parse straight from the raw blob.
    let mut from_bytes_out = Vec::new();
    {
        let mut compressor = FrameCompressor::new(super::CompressionLevel::Fastest);
        compressor
            .set_dictionary_from_bytes(dict_raw)
            .expect("dictionary bytes should parse");
        compressor.set_source(payload.as_slice());
        compressor.set_drain(&mut from_bytes_out);
        compressor.compress();
    }

    // Path B: full decode (builds the decoder tables too), then attach for
    // encoding via the `Dictionary` setter.
    let full_decode = crate::decoding::Dictionary::decode_dict(dict_raw)
        .expect("dictionary bytes should fully decode");
    let mut full_decode_out = Vec::new();
    {
        let mut compressor = FrameCompressor::new(super::CompressionLevel::Fastest);
        compressor
            .set_dictionary(full_decode)
            .expect("full-decode dictionary should attach");
        compressor.set_source(payload.as_slice());
        compressor.set_drain(&mut full_decode_out);
        compressor.compress();
    }

    assert_eq!(
        from_bytes_out, full_decode_out,
        "encoder-only dict parse must produce byte-identical output to the full decode"
    );
}

/// A dictionary with no ID is a raw-content one: the blob has no header to
/// carry an ID, and the frames built from it record none, so the decoder has to
/// be given the same bytes. It attaches like any other dictionary — what it
/// cannot do is be looked up by ID, which the registration path still enforces.
#[test]
fn set_dictionary_accepts_a_dictionary_without_an_id() {
    let raw_content = crate::decoding::Dictionary {
        id: 0,
        fse: crate::decoding::scratch::FSEScratch::new(),
        huf: crate::decoding::scratch::HuffmanScratch::new(),
        dict_content: vec![1, 2, 3],
        offset_hist: [1, 4, 8],
    };

    let mut compressor: FrameCompressor<
        &[u8],
        Vec<u8>,
        crate::encoding::match_generator::MatchGeneratorDriver,
    > = FrameCompressor::new(super::CompressionLevel::Fastest);
    compressor
        .set_dictionary(raw_content)
        .expect("a dictionary without an id must attach");

    let mut decoder = crate::decoding::FrameDecoder::new();
    let registered = decoder.add_dict(crate::decoding::Dictionary {
        id: 0,
        fse: crate::decoding::scratch::FSEScratch::new(),
        huf: crate::decoding::scratch::HuffmanScratch::new(),
        dict_content: vec![1, 2, 3],
        offset_hist: [1, 4, 8],
    });
    assert!(
        registered.is_err(),
        "registration keys on the id, so a zero one still has no slot"
    );
}

/// Upstream loads a dictionary buffer in `ZSTD_dct_auto` mode
/// (`ZSTD_CCtx_loadDictionary` -> `ZSTD_compress_insertDictionary`,
/// zstd_compress.c:5216-5222): a buffer whose first four bytes are not
/// `ZSTD_MAGIC_DICTIONARY` is raw content, not a malformed dictionary. Any file
/// can therefore be handed to `zstd -D`, and the same has to hold here, or a
/// caller that works against libzstd fails against this one.
#[test]
fn set_dictionary_from_bytes_takes_unmagicked_bytes_as_raw_content() {
    // Record-shaped, so the payload below actually matches into it: a blob the
    // encoder cannot use would round-trip even with the dictionary silently
    // dropped, and the frame-size check at the end would not hold.
    let raw_dict = b"tenant=demo table=orders op=put value=aaaaabbbbbcccccdddddeeeee\n".repeat(16);
    assert_ne!(
        &raw_dict[..4],
        &crate::decoding::DICTIONARY_MAGIC,
        "the fixture must not start with the dictionary magic",
    );
    let payload = b"tenant=demo table=orders op=put value=aaaaabbbbbcccccdddddeeeee\n".repeat(4);

    let mut with_dict = Vec::new();
    let mut compressor = FrameCompressor::new(super::CompressionLevel::Default);
    compressor
        .set_dictionary_from_bytes(&raw_dict)
        .expect("raw content must load the way `zstd -D` loads it");
    compressor.set_source(payload.as_slice());
    compressor.set_drain(&mut with_dict);
    compressor.compress();

    // A raw-content dictionary has no header to carry an id, so the frame
    // records none (upstream: `dictID = 0` is not written).
    let (header, _) = crate::decoding::frame::read_frame_header(with_dict.as_slice())
        .expect("the frame header should read back");
    assert_eq!(
        header.dictionary_id(),
        None,
        "a raw-content dictionary has no id to advertise",
    );

    let handle = crate::decoding::Dictionary::from_raw_content(0, raw_dict.clone())
        .expect("raw content is a valid dictionary")
        .into_handle();
    let mut decoded = vec![0u8; payload.len()];
    let written = FrameDecoder::new()
        .decode_all_with_dict_handle(&with_dict, &mut decoded, &handle)
        .expect("the frame decodes against the same bytes");
    assert_eq!(&decoded[..written], payload.as_slice());

    // And the dictionary was worth attaching: the same payload without it is
    // larger, which is what proves the content was primed rather than parsed
    // and discarded.
    let mut without_dict = Vec::new();
    let mut plain = FrameCompressor::new(super::CompressionLevel::Default);
    plain.set_source(payload.as_slice());
    plain.set_drain(&mut without_dict);
    plain.compress();
    assert!(
        with_dict.len() < without_dict.len(),
        "the raw content should have been primed: {} vs {} bytes without it",
        with_dict.len(),
        without_dict.len(),
    );
}

/// An empty buffer is how `ZSTD_CCtx_loadDictionary` is told there is no
/// dictionary: it clears whatever was attached and succeeds
/// (`ZSTD_clearAllDicts` then `return 0`, zstd_compress.c:1293-1295). Ours
/// refused it, so a caller could neither say "no dictionary" nor undo an
/// earlier one through the setter.
#[test]
fn set_dictionary_from_bytes_with_an_empty_buffer_clears_the_dictionary() {
    let raw_dict = b"tenant=demo table=orders op=put value=aaaaabbbbbcccccdddddeeeee\n".repeat(16);
    let payload = b"tenant=demo table=orders op=put value=aaaaabbbbbcccccdddddeeeee\n".repeat(4);

    let mut compressor: FrameCompressor<&[u8], Vec<u8>> =
        FrameCompressor::new(super::CompressionLevel::Default);
    compressor
        .set_dictionary_from_bytes(&raw_dict)
        .expect("the dictionary attaches");
    let cleared = compressor
        .set_dictionary_from_bytes(&[])
        .expect("an empty buffer is how a caller says there is no dictionary");
    assert!(
        cleared.is_some(),
        "clearing hands back the dictionary that was attached",
    );

    // What it compresses is now what a compressor that never saw a dictionary
    // compresses — the attach is gone, not merely emptied.
    let mut after_clear = Vec::new();
    let mut c = FrameCompressor::new(super::CompressionLevel::Default);
    c.set_dictionary_from_bytes(&raw_dict).expect("attach");
    c.set_dictionary_from_bytes(&[]).expect("clear");
    c.set_source(payload.as_slice());
    c.set_drain(&mut after_clear);
    c.compress();

    let mut never_had_one = Vec::new();
    let mut plain = FrameCompressor::new(super::CompressionLevel::Default);
    plain.set_source(payload.as_slice());
    plain.set_drain(&mut never_had_one);
    plain.compress();

    assert_eq!(
        after_clear, never_had_one,
        "a cleared dictionary must leave the frame a plain one",
    );
}

/// Taking either kind is not taking anything: a blob that claims to be a
/// serialized dictionary by carrying the magic, and then does not parse, is a
/// corrupt dictionary and must be refused rather than quietly re-read as raw
/// content (upstream classifies on the magic alone and then fails the parse).
#[test]
fn set_dictionary_from_bytes_rejects_a_corrupt_serialized_dictionary() {
    let mut corrupt = crate::decoding::DICTIONARY_MAGIC.to_vec();
    corrupt.extend_from_slice(&[0xFF; 60]);

    let mut compressor: FrameCompressor<
        &[u8],
        Vec<u8>,
        crate::encoding::match_generator::MatchGeneratorDriver,
    > = FrameCompressor::new(super::CompressionLevel::Fastest);
    assert!(
        compressor.set_dictionary_from_bytes(&corrupt).is_err(),
        "a magic-prefixed blob that does not parse is corrupt, not raw content",
    );
}

#[test]
fn set_dictionary_rejects_zero_repeat_offsets() {
    let invalid = crate::decoding::Dictionary {
        id: 1,
        fse: crate::decoding::scratch::FSEScratch::new(),
        huf: crate::decoding::scratch::HuffmanScratch::new(),
        dict_content: vec![1, 2, 3],
        offset_hist: [0, 4, 8],
    };

    let mut compressor: FrameCompressor<
        &[u8],
        Vec<u8>,
        crate::encoding::match_generator::MatchGeneratorDriver,
    > = FrameCompressor::new(super::CompressionLevel::Fastest);
    let result = compressor.set_dictionary(invalid);
    assert!(matches!(
        result,
        Err(
            crate::decoding::errors::DictionaryDecodeError::ZeroRepeatOffsetInDictionary {
                index: 0
            }
        )
    ));
}

#[test]
fn uncompressed_mode_does_not_require_dictionary() {
    let dict_id = 0xABCD_0001;
    let dict = crate::decoding::Dictionary::from_raw_content(dict_id, b"shared-history".to_vec())
        .expect("raw dictionary should be valid");

    let payload = b"plain-bytes-that-should-stay-raw";
    let mut output = Vec::new();
    let mut compressor = FrameCompressor::new(super::CompressionLevel::Uncompressed);
    compressor
        .set_dictionary(dict)
        .expect("dictionary should attach in uncompressed mode");
    compressor.set_source(payload.as_slice());
    compressor.set_drain(&mut output);
    compressor.compress();

    let (frame_header, _) = crate::decoding::frame::read_frame_header(output.as_slice())
        .expect("encoded frame should have a header");
    assert_eq!(
        frame_header.dictionary_id(),
        None,
        "raw/uncompressed frames must not advertise dictionary dependency"
    );

    let mut decoder = FrameDecoder::new();
    let mut decoded = Vec::with_capacity(payload.len());
    decoder.decode_all_to_vec(&output, &mut decoded).unwrap();
    assert_eq!(decoded, payload);
}

#[test]
fn default_level_tiny_raw_dict_compresses_cleanly() {
    // Coverage for the dfast dict-attach fast path with a
    // sub-min-match raw-content dictionary: the dict-table probe in
    // `start_matching_fast_loop` is gated on the dict table actually
    // existing (`table().is_some()`), not merely on `is_attached()`,
    // so a dictionary whose hashable region is shorter than the
    // short-hash lookahead (where `prime_dict_tables_for_range`
    // returns before allocating the tables) never dereferences a
    // null dict pointer. Compressing at the default (dfast) level
    // with such a dict must succeed.
    let dict_id = 0xABCD_0009;
    let dict = crate::decoding::Dictionary::from_raw_content(dict_id, b"abc".to_vec())
        .expect("raw dictionary should be valid");
    let payload = b"the quick brown fox jumps over the lazy dog, repeatedly and at length";
    let mut output = Vec::new();
    let mut compressor = FrameCompressor::new(super::CompressionLevel::Default);
    compressor
        .set_dictionary(dict)
        .expect("tiny raw dictionary should attach");
    compressor.set_source(payload.as_slice());
    compressor.set_drain(&mut output);
    compressor.compress();
    assert!(!output.is_empty(), "compression should produce a frame");

    // The emitted frame must advertise the attached dictionary id, proving
    // the tiny-dict path stayed active (the payload round-trips either way,
    // so without this the test would also pass on a silent no-dict frame).
    let (frame_header, _) = crate::decoding::frame::read_frame_header(output.as_slice())
        .expect("encoded frame should have a readable header");
    assert_eq!(
        frame_header.dictionary_id(),
        Some(dict_id),
        "tiny raw dict frame should still advertise its dictionary id",
    );

    // Full roundtrip: decode the dict-compressed frame with the SAME
    // dictionary attached and confirm byte-exact recovery — proves the
    // tiny-dict fast path produces a correct frame, not just a non-empty
    // one.
    let decode_dict = crate::decoding::Dictionary::from_raw_content(dict_id, b"abc".to_vec())
        .expect("raw dictionary should be valid");
    let mut decoder = FrameDecoder::new();
    decoder
        .add_dict(decode_dict)
        .expect("decoder dict should attach");
    let mut decoded = Vec::with_capacity(payload.len());
    decoder
        .decode_all_to_vec(&output, &mut decoded)
        .expect("dict roundtrip should decode");
    assert_eq!(decoded, payload, "tiny-dict roundtrip mismatch");
}

/// Exercises the dictionary dual-probe (live + immutable dict tables)
/// in the Fast / dfast / Row match finders with a dict whose content
/// the payload actually reuses, so each backend's dict long/short
/// probe (and the dfast `ip+1` dict-long retry) is reached and the
/// dict-compressed frame round-trips through a decoder primed with the
/// same dict. The 3-byte-dict test above only proves the null-table
/// guard; this proves the full attach path produces correct frames.
#[test]
fn dict_attach_roundtrips_across_backends_with_matching_payload() {
    let dict_id = 0xD1C7_0001;
    // Distinct lines so the payload does NOT self-compress: each line
    // appears exactly once in the payload, so without the dictionary there
    // are no in-frame back-references to exploit. The dictionary holds the
    // SAME lines, so the only way the output shrinks is if the dict probe
    // actually fires. A no-dict baseline below pins that the dict path ran
    // (self-compressible payloads would round-trip + stay small via
    // in-frame matches alone, proving nothing).
    let line = |i: u32| {
        alloc::format!(
            "ts=2026-03-26T21:{:02}:{:02}Z level=INFO msg=\"event {i:05}\" tenant=t{i} region=eu\n",
            i / 60 % 60,
            i % 60,
        )
        .into_bytes()
    };
    let mut dict_content = Vec::new();
    for i in 0..256u32 {
        dict_content.extend_from_slice(&line(i));
    }
    // Payload = the same distinct lines in a different (stride) order, each
    // once → no self-repeats, every line is a dictionary match.
    let mut payload = Vec::new();
    let mut i = 0u32;
    for _ in 0..256u32 {
        payload.extend_from_slice(&line(i));
        i = (i + 97) % 256; // coprime stride → permutation, no adjacency
    }

    let compress_at = |level, dict: Option<Vec<u8>>| -> Vec<u8> {
        let mut compressor = FrameCompressor::new(level);
        if let Some(bytes) = dict {
            let d = crate::decoding::Dictionary::from_raw_content(dict_id, bytes)
                .expect("raw dictionary should be valid");
            compressor
                .set_dictionary(d)
                .expect("dictionary should attach");
        }
        let mut out = Vec::new();
        compressor.set_source(payload.as_slice());
        compressor.set_drain(&mut out);
        compressor.compress();
        out
    };

    for level in [
        super::CompressionLevel::Level(-5), // Fast (negative)
        super::CompressionLevel::Level(1),  // Fast
        super::CompressionLevel::Default,   // dfast (L3)
        super::CompressionLevel::Level(8),  // Row-backed lazy2
    ] {
        let out = compress_at(level, Some(dict_content.clone()));
        let no_dict = compress_at(level, None);
        // The dict path MUST measurably beat no-dict on this
        // non-self-compressible payload — otherwise the dict probe never
        // fired and the roundtrip below would prove nothing.
        assert!(
            out.len() < no_dict.len(),
            "level {level:?}: dict-primed output ({}) must beat no-dict ({}) — dict probe did not fire",
            out.len(),
            no_dict.len(),
        );

        let ddict = crate::decoding::Dictionary::from_raw_content(dict_id, dict_content.clone())
            .expect("raw dictionary should be valid");
        let mut decoder = FrameDecoder::new();
        decoder.add_dict(ddict).expect("decoder dict should attach");
        let mut decoded = Vec::with_capacity(payload.len());
        decoder
            .decode_all_to_vec(&out, &mut decoded)
            .unwrap_or_else(|e| panic!("level {level:?}: dict roundtrip decode failed: {e:?}"));
        assert_eq!(decoded, payload, "level {level:?}: dict roundtrip mismatch");
    }
}

/// Reusing one compressor across independent frames with DIFFERENT
/// dictionaries must drop the per-backend dict cache on each swap
/// (Simple/Dfast/Row keep the attach index across frames). Without the
/// invalidation a later frame would reuse the previous dict's rows.
/// Each frame round-trips through a decoder primed with its own dict.
#[test]
fn dict_swap_across_reused_compressor_roundtrips() {
    // Distinct lines per dict (not a single repeated line) so payloads do
    // NOT self-compress: each line appears once, so a frame only shrinks if
    // the dict probe fires, and — crucially for the invalidation check — if
    // frame B reused dict A's stale rows it would emit offsets into A's
    // distinct content, which decode under dict B reconstructs as WRONG
    // bytes (caught by the roundtrip). A single repeated line would hide
    // pollution behind in-frame matches.
    let lines = |tag: &str| -> (Vec<u8>, Vec<u8>) {
        let line = |i: u32| alloc::format!("{tag} record {i:05} field=value{i} end\n").into_bytes();
        let mut dict = Vec::new();
        for i in 0..256u32 {
            dict.extend_from_slice(&line(i));
        }
        let mut payload = Vec::new();
        let mut i = 0u32;
        for _ in 0..256u32 {
            payload.extend_from_slice(&line(i));
            i = (i + 97) % 256;
        }
        (dict, payload)
    };
    let (dict_a, payload_a) = lines("alpha");
    let (dict_b, payload_b) = lines("bravo");

    for level in [
        super::CompressionLevel::Default,
        super::CompressionLevel::Level(8),
    ] {
        let no_dict = |payload: &[u8]| -> usize {
            let mut c: FrameCompressor = FrameCompressor::new(level);
            c.compress_independent_frame(payload).len()
        };
        let no_dict_a = no_dict(&payload_a);
        let no_dict_b = no_dict(&payload_b);

        let mut compressor: FrameCompressor = FrameCompressor::new(level);
        for (dict_bytes, payload, no_dict_len) in [
            (&dict_a, &payload_a, no_dict_a),
            (&dict_b, &payload_b, no_dict_b),
        ] {
            let dict =
                crate::decoding::Dictionary::from_raw_content(0xD1C7_0002, dict_bytes.clone())
                    .expect("raw dictionary should be valid");
            compressor
                .set_dictionary(dict)
                .expect("dictionary should attach");
            let out = compressor.compress_independent_frame(payload.as_slice());
            assert!(
                out.len() < no_dict_len,
                "level {level:?}: dict frame ({}) must beat no-dict ({}) — dict probe did not fire",
                out.len(),
                no_dict_len,
            );

            let ddict =
                crate::decoding::Dictionary::from_raw_content(0xD1C7_0002, dict_bytes.clone())
                    .expect("raw dictionary should be valid");
            let mut decoder = FrameDecoder::new();
            decoder.add_dict(ddict).expect("decoder dict should attach");
            let mut decoded = Vec::with_capacity(payload.len());
            decoder
                .decode_all_to_vec(&out, &mut decoded)
                .unwrap_or_else(|e| panic!("level {level:?}: dict-swap decode failed: {e:?}"));
            assert_eq!(
                decoded, *payload,
                "level {level:?}: dict-swap roundtrip mismatch (stale dict rows?)"
            );
        }
    }
}

#[test]
fn dictionary_roundtrip_stays_valid_after_output_exceeds_window() {
    use crate::encoding::match_generator::MatchGeneratorDriver;

    let dict_id = 0xABCD_0002;
    let dict = crate::decoding::Dictionary::from_raw_content(dict_id, b"abcdefgh".to_vec())
        .expect("raw dictionary should be valid");
    let dict_for_decoder =
        crate::decoding::Dictionary::from_raw_content(dict_id, b"abcdefgh".to_vec())
            .expect("raw dictionary should be valid");

    // Payload must exceed the encoder's advertised window (512 KiB
    // for Fastest after `window_log = 19` alignment with upstream zstd's
    // L1 fast row in `clevels.h`) so the test actually exercises
    // cross-window-boundary behavior.
    let payload = b"abcdefgh".repeat(512 * 1024 / 8 + 64);
    let matcher = MatchGeneratorDriver::new(1024, 1);

    let mut no_dict_output = Vec::new();
    let mut no_dict_compressor =
        FrameCompressor::new_with_matcher(matcher, super::CompressionLevel::Fastest);
    no_dict_compressor.set_source(payload.as_slice());
    no_dict_compressor.set_drain(&mut no_dict_output);
    no_dict_compressor.compress();
    let (no_dict_frame_header, _) =
        crate::decoding::frame::read_frame_header(no_dict_output.as_slice())
            .expect("baseline frame should have a header");
    let no_dict_window = no_dict_frame_header
        .window_size()
        .expect("window size should be present");

    let mut output = Vec::new();
    let matcher = MatchGeneratorDriver::new(1024, 1);
    let mut compressor =
        FrameCompressor::new_with_matcher(matcher, super::CompressionLevel::Fastest);
    compressor
        .set_dictionary(dict)
        .expect("dictionary should attach");
    compressor.set_source(payload.as_slice());
    compressor.set_drain(&mut output);
    compressor.compress();

    let (frame_header, _) = crate::decoding::frame::read_frame_header(output.as_slice())
        .expect("encoded frame should have a header");
    let advertised_window = frame_header
        .window_size()
        .expect("window size should be present");
    assert_eq!(
        advertised_window, no_dict_window,
        "dictionary priming must not inflate advertised window size"
    );
    assert!(
        payload.len() > advertised_window as usize,
        "test must cross the advertised window boundary"
    );

    let mut decoder = FrameDecoder::new();
    decoder.add_dict(dict_for_decoder).unwrap();
    let mut decoded = Vec::with_capacity(payload.len());
    decoder.decode_all_to_vec(&output, &mut decoded).unwrap();
    assert_eq!(decoded, payload);
}

#[test]
fn source_size_hint_with_dictionary_keeps_roundtrip_and_nonincreasing_window() {
    let dict_id = 0xABCD_0004;
    let dict_content = b"abcd".repeat(1024); // 4 KiB dictionary history
    let dict = crate::decoding::Dictionary::from_raw_content(dict_id, dict_content).unwrap();
    let dict_for_decoder =
        crate::decoding::Dictionary::from_raw_content(dict_id, b"abcd".repeat(1024)).unwrap();
    let payload = b"abcdabcdabcdabcd".repeat(128);

    let mut hinted_output = Vec::new();
    let mut hinted = FrameCompressor::new(super::CompressionLevel::Fastest);
    hinted.set_dictionary(dict).unwrap();
    hinted.set_source_size_hint(1);
    hinted.set_source(payload.as_slice());
    hinted.set_drain(&mut hinted_output);
    hinted.compress();

    let mut no_hint_output = Vec::new();
    let mut no_hint = FrameCompressor::new(super::CompressionLevel::Fastest);
    no_hint
        .set_dictionary(
            crate::decoding::Dictionary::from_raw_content(dict_id, b"abcd".repeat(1024)).unwrap(),
        )
        .unwrap();
    no_hint.set_source(payload.as_slice());
    no_hint.set_drain(&mut no_hint_output);
    no_hint.compress();

    let hinted_window = crate::decoding::frame::read_frame_header(hinted_output.as_slice())
        .expect("encoded frame should have a header")
        .0
        .window_size()
        .expect("window size should be present");
    let no_hint_window = crate::decoding::frame::read_frame_header(no_hint_output.as_slice())
        .expect("encoded frame should have a header")
        .0
        .window_size()
        .expect("window size should be present");
    assert!(
        hinted_window <= no_hint_window,
        "source-size hint should not increase advertised window with dictionary priming",
    );

    let mut decoder = FrameDecoder::new();
    decoder.add_dict(dict_for_decoder).unwrap();
    let mut decoded = Vec::with_capacity(payload.len());
    decoder
        .decode_all_to_vec(&hinted_output, &mut decoded)
        .unwrap();
    assert_eq!(decoded, payload);
}

/// A dictionary segment embedded ONCE in otherwise-incompressible
/// input must be matched against the dictionary. Before the fix the
/// raw-fast-path (which skips matching) fired on the
/// incompressible-looking block and the dictionary was never searched,
/// so `with_dict` came out the same size as `no_dict` (the embedded
/// match was lost). Now the block compresses against the dict.
#[test]
fn dictionary_segment_in_incompressible_input_is_matched() {
    // Deterministic LCG bytes: high-entropy, so the only compressible
    // content is the embedded dictionary segment.
    fn lcg(seed: u64, n: usize) -> alloc::vec::Vec<u8> {
        let mut s = seed;
        (0..n)
            .map(|_| {
                s = s
                    .wrapping_mul(6364136223846793005)
                    .wrapping_add(1442695040888963407);
                (s >> 56) as u8
            })
            .collect()
    }
    let dict_id = 0x00DC_7777;
    let r = lcg(1, 512); // the dictionary content
    let mut payload = lcg(2, 2000); // incompressible filler before
    payload.extend_from_slice(&r); // the single dict-matchable segment
    payload.extend_from_slice(&lcg(3, 1500)); // filler after

    // Precondition: the payload must actually look incompressible so
    // that the raw-fast-path WOULD fire (and skip matching) without
    // the fix. If the heuristic ever changes and this no longer holds,
    // the test below would pass vacuously — assert it up front.
    assert!(
        crate::encoding::incompressible::block_looks_incompressible(&payload),
        "test payload must look incompressible to exercise the raw-fast-path",
    );

    let compress = |level: super::CompressionLevel, dict: Option<&[u8]>| -> alloc::vec::Vec<u8> {
        let mut out = alloc::vec::Vec::new();
        let mut c = FrameCompressor::new(level);
        if let Some(d) = dict {
            c.set_dictionary(
                crate::decoding::Dictionary::from_raw_content(dict_id, d.to_vec()).unwrap(),
            )
            .unwrap();
        }
        c.set_source(payload.as_slice());
        c.set_drain(&mut out);
        c.compress();
        out
    };

    for lvl in [
        super::CompressionLevel::Level(2),
        super::CompressionLevel::Level(6),
        super::CompressionLevel::Level(19),
    ] {
        let with_dict = compress(lvl, Some(&r));
        let no_dict = compress(lvl, None);
        // The 512-byte dict segment should be matched, saving most of
        // its length (generous slack for sequence/header coding).
        assert!(
            with_dict.len() + 300 < no_dict.len(),
            "{lvl:?}: dict segment not matched (with_dict={}, no_dict={})",
            with_dict.len(),
            no_dict.len(),
        );
        // The dict-compressed frame must round-trip through the decoder.
        let mut decoder = FrameDecoder::new();
        decoder
            .add_dict(crate::decoding::Dictionary::from_raw_content(dict_id, r.clone()).unwrap())
            .unwrap();
        let mut decoded = Vec::with_capacity(payload.len());
        decoder.decode_all_to_vec(&with_dict, &mut decoded).unwrap();
        assert_eq!(decoded, payload, "{lvl:?}: dict round-trip mismatch");

        // A dictionary that does NOT appear in the input must not make
        // the output larger than the no-dict (raw) encoding: the
        // post-compress raw fallback covers incompressible-with-dict.
        let unrelated = lcg(99, 512);
        let with_bad_dict = compress(lvl, Some(&unrelated));
        assert!(
            with_bad_dict.len() <= no_dict.len() + 16,
            "{lvl:?}: unhelpful dict expanded output (with={}, no_dict={})",
            with_bad_dict.len(),
            no_dict.len(),
        );
    }
}

#[test]
fn source_size_hint_with_dictionary_keeps_roundtrip_for_larger_payload() {
    let dict_id = 0xABCD_0005;
    let dict_content = b"abcd".repeat(1024); // 4 KiB dictionary history
    let dict = crate::decoding::Dictionary::from_raw_content(dict_id, dict_content).unwrap();
    let dict_for_decoder =
        crate::decoding::Dictionary::from_raw_content(dict_id, b"abcd".repeat(1024)).unwrap();
    let payload = b"abcd".repeat(1024); // 4 KiB payload
    let payload_len = payload.len() as u64;

    let mut hinted_output = Vec::new();
    let mut hinted = FrameCompressor::new(super::CompressionLevel::Fastest);
    hinted.set_dictionary(dict).unwrap();
    hinted.set_source_size_hint(payload_len);
    hinted.set_source(payload.as_slice());
    hinted.set_drain(&mut hinted_output);
    hinted.compress();

    let mut no_hint_output = Vec::new();
    let mut no_hint = FrameCompressor::new(super::CompressionLevel::Fastest);
    no_hint
        .set_dictionary(
            crate::decoding::Dictionary::from_raw_content(dict_id, b"abcd".repeat(1024)).unwrap(),
        )
        .unwrap();
    no_hint.set_source(payload.as_slice());
    no_hint.set_drain(&mut no_hint_output);
    no_hint.compress();

    let hinted_window = crate::decoding::frame::read_frame_header(hinted_output.as_slice())
        .expect("encoded frame should have a header")
        .0
        .window_size()
        .expect("window size should be present");
    let no_hint_window = crate::decoding::frame::read_frame_header(no_hint_output.as_slice())
        .expect("encoded frame should have a header")
        .0
        .window_size()
        .expect("window size should be present");
    assert!(
        hinted_window <= no_hint_window,
        "source-size hint should not increase advertised window with dictionary priming",
    );

    let mut decoder = FrameDecoder::new();
    decoder.add_dict(dict_for_decoder).unwrap();
    let mut decoded = Vec::with_capacity(payload.len());
    decoder
        .decode_all_to_vec(&hinted_output, &mut decoded)
        .unwrap();
    assert_eq!(decoded, payload);
}

#[test]
fn custom_matcher_without_dictionary_priming_does_not_advertise_dict_id() {
    let dict_id = 0xABCD_0003;
    let dict = crate::decoding::Dictionary::from_raw_content(dict_id, b"abcdefgh".to_vec())
        .expect("raw dictionary should be valid");
    let payload = b"abcdefghabcdefgh";

    let mut output = Vec::new();
    let matcher = NoDictionaryMatcher::new(64);
    let mut compressor =
        FrameCompressor::new_with_matcher(matcher, super::CompressionLevel::Fastest);
    compressor
        .set_dictionary(dict)
        .expect("dictionary should attach");
    compressor.set_source(payload.as_slice());
    compressor.set_drain(&mut output);
    compressor.compress();

    let (frame_header, _) = crate::decoding::frame::read_frame_header(output.as_slice())
        .expect("encoded frame should have a header");
    assert_eq!(
        frame_header.dictionary_id(),
        None,
        "matchers that do not support dictionary priming must not advertise dictionary dependency"
    );

    let mut decoder = FrameDecoder::new();
    let mut decoded = Vec::with_capacity(payload.len());
    decoder.decode_all_to_vec(&output, &mut decoded).unwrap();
    assert_eq!(decoded, payload);
}

#[cfg(feature = "hash")]
#[test]
fn checksum_two_frames_reused_compressor() {
    // Compress the same data twice using the same compressor and verify that:
    // 1. The checksum written in each frame matches what the decoder calculates.
    // 2. The hasher is correctly reset between frames (no cross-contamination).
    //    If the hasher were NOT reset, the second frame's calculated checksum
    //    would differ from the one stored in the frame data, causing assert_eq to fail.
    let data: Vec<u8> = (0u8..=255).cycle().take(1024).collect();

    let mut compressor = FrameCompressor::new(super::CompressionLevel::Uncompressed);

    // --- Frame 1 ---
    let mut compressed1 = Vec::new();
    compressor.set_source(data.as_slice());
    compressor.set_drain(&mut compressed1);
    compressor.compress();

    // --- Frame 2 (reuse the same compressor) ---
    let mut compressed2 = Vec::new();
    compressor.set_source(data.as_slice());
    compressor.set_drain(&mut compressed2);
    compressor.compress();

    fn decode_and_collect(compressed: &[u8]) -> (Vec<u8>, Option<u32>, Option<u32>) {
        let mut decoder = FrameDecoder::new();
        let mut source = compressed;
        decoder.reset(&mut source).unwrap();
        while !decoder.is_finished() {
            decoder
                .decode_blocks(&mut source, crate::decoding::BlockDecodingStrategy::All)
                .unwrap();
        }
        let mut decoded = Vec::new();
        decoder.collect_to_writer(&mut decoded).unwrap();
        (
            decoded,
            decoder.get_checksum_from_data(),
            decoder.get_calculated_checksum(),
        )
    }

    let (decoded1, chksum_from_data1, chksum_calculated1) = decode_and_collect(&compressed1);
    assert_eq!(decoded1, data, "frame 1: decoded data mismatch");
    assert_eq!(
        chksum_from_data1, chksum_calculated1,
        "frame 1: checksum mismatch"
    );

    let (decoded2, chksum_from_data2, chksum_calculated2) = decode_and_collect(&compressed2);
    assert_eq!(decoded2, data, "frame 2: decoded data mismatch");
    assert_eq!(
        chksum_from_data2, chksum_calculated2,
        "frame 2: checksum mismatch"
    );

    // Same data compressed twice must produce the same checksum.
    // If state leaked across frames, the second calculated checksum would differ.
    assert_eq!(
        chksum_from_data1, chksum_from_data2,
        "frame 1 and frame 2 should have the same checksum (same data, hash must reset per frame)"
    );
}

#[cfg(feature = "lsm")]
#[test]
fn frame_emit_info_decompressed_ranges_match_decoded_output() {
    // Part A correctness: the per-block `decompressed_size` captured during
    // encode (and the `decompressed_byte_range` prefix sum derived from it)
    // must describe the real decoded output exactly — one entry per
    // physical block, contiguous, summing to the full decompressed length.
    // A multi-block compressible payload exercises the Compressed-block
    // path (whose regenerated size is NOT on the wire, so it relies on the
    // encode-side capture this test guards).
    let data = emit_info_fixture_data();

    // Cover both the single-block-per-chunk path (Default) and the
    // Level(16..=22) post-split path (multiple physical partitions per
    // input chunk), since lsm-tree compresses at zstd:22 and post-split
    // is the riskiest capture site (per-partition `src_size`).
    for level in [
        super::CompressionLevel::Default,
        super::CompressionLevel::Level(22),
    ] {
        let mut compressed = Vec::new();
        let mut compressor = FrameCompressor::new(level);
        // Pledge the source size so the high-level (22) window shrinks to
        // fit the payload, keeping the frame compact (no oversized window
        // descriptor for a small input). Still >= 128 KiB, so post-split
        // eligibility is preserved.
        compressor.set_source_size_hint(data.len() as u64);
        compressor.set_source(data.as_slice());
        compressor.set_drain(&mut compressed);
        compressor.compress();

        let info = compressor
            .last_frame_emit_info()
            .expect("emit info populated after compress")
            .clone();

        // Reference: full decode of the same frame.
        let mut decoder = FrameDecoder::new();
        let mut source = compressed.as_slice();
        decoder.reset(&mut source).unwrap();
        while !decoder.is_finished() {
            decoder
                .decode_blocks(&mut source, crate::decoding::BlockDecodingStrategy::All)
                .unwrap();
        }
        let mut decoded = Vec::new();
        decoder.collect_to_writer(&mut decoded).unwrap();
        assert_eq!(decoded, data, "sanity: frame must round-trip ({level:?})");

        assert!(
            info.blocks.len() >= 2,
            "fixture must span multiple blocks to exercise the mapping ({level:?}, got {})",
            info.blocks.len()
        );
        assert!(
            info.blocks.last().unwrap().last_block,
            "final block must carry last_block ({level:?})"
        );

        // Pin the Level(22) post-split path: the owned loop feeds the
        // encoder MAX_BLOCK_SIZE input chunks, so without post-split the
        // block count cannot exceed the chunk count. More blocks than
        // chunks proves at least one chunk was split into multiple physical
        // partitions (the per-partition `src_size` capture under test).
        if matches!(level, super::CompressionLevel::Level(22)) {
            let max_block = crate::common::MAX_BLOCK_SIZE as usize;
            let n_chunks = data.len().div_ceil(max_block);
            assert!(
                info.blocks.len() > n_chunks,
                "Level(22) must exercise post-split: {} blocks for {} input chunks",
                info.blocks.len(),
                n_chunks
            );
        }

        // Per-block ranges: contiguous, zero-based, summing to the full output.
        let mut expected_start = 0u64;
        for i in 0..info.blocks.len() {
            let range = info
                .decompressed_byte_range(i)
                .expect("in-bounds block has a range");
            assert_eq!(
                range.start, expected_start,
                "block {i} range must start where the previous ended ({level:?})"
            );
            assert_eq!(
                u64::from(info.blocks[i].decompressed_size),
                range.end - range.start,
                "block {i} decompressed_size must equal its range width ({level:?})"
            );
            // Validate the mapping against REAL per-block bytes, not just
            // prefix-sum consistency: decode block `i` alone and require it
            // to equal the corresponding slice of the full decode. A
            // sidecar that swapped sizes between adjacent blocks (same sum,
            // same contiguity) would fail here.
            let mut psrc = compressed.as_slice();
            let mut pdec = FrameDecoder::new();
            pdec.reset(&mut psrc).unwrap();
            let pd = pdec
                .decode_blocks_partial(&mut psrc, i as u32, i as u32 + 1, None, false)
                .unwrap();
            assert!(
                pd.stopped_at.is_none(),
                "block {i} must decode cleanly ({level:?})"
            );
            assert_eq!(
                pd.data.as_slice(),
                &decoded[range.start as usize..range.end as usize],
                "block {i} partial-decode bytes must equal the full-decode slice ({level:?})"
            );
            expected_start = range.end;
        }
        assert_eq!(
            expected_start,
            decoded.len() as u64,
            "block decompressed sizes must sum to the full decoded length ({level:?})"
        );
        assert_eq!(
            info.decompressed_byte_range(info.blocks.len()),
            None,
            "out-of-range index yields None ({level:?})"
        );
    }
}

/// ~400 KiB semi-repetitive payload (long runs interleaved with a stride
/// phrase) that compresses into several multi-block frames across levels.
#[cfg(feature = "lsm")]
fn emit_info_fixture_data() -> Vec<u8> {
    let mut data: Vec<u8> = Vec::with_capacity(400 * 1024);
    let mut x = 0x9E37_79B9u32;
    while data.len() < 400 * 1024 {
        x ^= x << 13;
        x ^= x >> 17;
        x ^= x << 5;
        let run = 16 + (x as usize % 48);
        let byte = (x >> 24) as u8;
        for _ in 0..run {
            data.push(byte);
        }
        data.extend_from_slice(b"the quick brown fox jumps over the lazy dog\n");
    }
    data
}

#[cfg(feature = "lsm")]
#[test]
fn frame_emit_info_decompressed_ranges_match_on_borrowed_oneshot_path() {
    // The borrowed one-shot path (`compress_independent_frame` ->
    // `run_borrowed_block_loop` -> `compress_block_encoded_borrowed`)
    // threads the decompressed-size sidecar through a DIFFERENT emit site
    // than the owned/streaming loop, so it needs its own per-block mapping
    // check. A Fast level keeps the encoder on the borrowed-eligible
    // (Simple matcher) path.
    let data = emit_info_fixture_data();

    let mut compressor: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Fastest);
    let compressed = compressor.compress_independent_frame(data.as_slice());
    let info = compressor
        .last_frame_emit_info()
        .expect("emit info populated after compress_independent_frame")
        .clone();
    // Pin the compressed-block path: without this the fixture could regress
    // into the raw-fast fallback and still pass via the Raw wire-size
    // fallback in populate_frame_emit_info, never exercising the borrowed
    // compressed-block sidecar capture this test targets.
    assert!(
        info.blocks
            .iter()
            .any(|b| matches!(b.block_type, crate::blocks::block::BlockType::Compressed)),
        "borrowed-path fixture must emit at least one compressed block"
    );
    assert!(
        info.blocks.len() >= 2,
        "borrowed fixture must span multiple blocks (got {})",
        info.blocks.len()
    );
    assert!(info.blocks.last().unwrap().last_block);

    // Full decode reference.
    let mut decoder = FrameDecoder::new();
    let mut source = compressed.as_slice();
    decoder.reset(&mut source).unwrap();
    while !decoder.is_finished() {
        decoder
            .decode_blocks(&mut source, crate::decoding::BlockDecodingStrategy::All)
            .unwrap();
    }
    let mut decoded = Vec::new();
    decoder.collect_to_writer(&mut decoded).unwrap();
    assert_eq!(decoded, data, "borrowed one-shot frame must round-trip");

    // Each block's mapping must match real per-block bytes.
    let mut expected_start = 0u64;
    for i in 0..info.blocks.len() {
        let range = info.decompressed_byte_range(i).unwrap();
        assert_eq!(range.start, expected_start, "block {i} range contiguity");
        let mut psrc = compressed.as_slice();
        let mut pdec = FrameDecoder::new();
        pdec.reset(&mut psrc).unwrap();
        let pd = pdec
            .decode_blocks_partial(&mut psrc, i as u32, i as u32 + 1, None, false)
            .unwrap();
        assert!(pd.stopped_at.is_none(), "block {i} must decode cleanly");
        assert_eq!(
            pd.data.as_slice(),
            &decoded[range.start as usize..range.end as usize],
            "borrowed block {i} partial-decode bytes must equal the full-decode slice"
        );
        expected_start = range.end;
    }
    assert_eq!(
        expected_start,
        decoded.len() as u64,
        "ranges sum to full length"
    );
}

// The fuzz-artifact interop replay (C-compress -> our-decode and
// our-compress -> C-decode) moved to `ffi-bench/tests/fuzz_interop.rs` so
// the library crate never links libzstd.

/// Homogeneous input — every byte the same — must NOT be split:
/// both border histograms are identical (all 512 hits on a single
/// slot), so `presplit_fingerprints_differ` returns `false` and the
/// function takes the early-return path at
/// `zstd_preSplit.c:214` returning `blockSize`.
#[test]
fn split_block_from_borders_keeps_homogeneous_block() {
    let block = vec![0xAAu8; MAX_BLOCK_SIZE as usize];
    let split = super::split_block_from_borders(&block);
    assert_eq!(split, MAX_BLOCK_SIZE as usize);
}

/// Heterogeneous input — first half all zeros, second half a
/// counter sequence — has clearly distinguishable border
/// histograms, so the borders heuristic decides to split.
///
/// The transition sits at exactly the block midpoint, so the
/// middle 512-byte sample (`block[mid-256..mid+256]`) is half
/// zeros + half counter values. That makes it roughly
/// equidistant from both border fingerprints — the
/// `abs_diff(dist_from_begin, dist_from_end) < min_distance`
/// branch fires and the heuristic returns the midpoint (64 KiB)
/// per `zstd_preSplit.c:222`. The test asserts the exact value
/// rather than just "one of {32K, 64K, 96K}" so a regression
/// to a different quantised arm cannot silently slip through.
#[test]
fn split_block_from_borders_returns_midpoint_for_centred_transition() {
    let mut block = vec![0u8; MAX_BLOCK_SIZE as usize];
    for (i, byte) in block
        .iter_mut()
        .enumerate()
        .skip(MAX_BLOCK_SIZE as usize / 2)
    {
        *byte = (i % 251 + 1) as u8;
    }
    let split = super::split_block_from_borders(&block);
    assert_eq!(
        split,
        64 * 1024,
        "centred-transition fixture must take the symmetric \
             midpoint arm (`abs_diff < min_distance`), got {split}"
    );
}

/// `level_pre_split` resolves the per-level split knob through the
/// `LevelParams` table: upstream's `splitLevels` up to lazy depth 1
/// (fast → 0, dfast → 1, greedy/lazy → 2), two steps coarser above (see
/// `pre_split_for` for why): lazy2/btlazy2 → 1 (byChunks rate 43),
/// btopt/btultra/btultra2 → 2 (rate 11). `Uncompressed` has no numeric level
/// so it stays `None`.
#[test]
fn pre_split_level_dispatches_by_compression_level() {
    use crate::encoding::CompressionLevel;
    use crate::encoding::levels::config::level_pre_split;
    assert_eq!(level_pre_split(CompressionLevel::Uncompressed), None);
    // Fastest = level 1 (fast) → 0 (from-borders).
    assert_eq!(level_pre_split(CompressionLevel::Fastest), Some(0));
    // Default = level 3 (dfast) → 1 (byChunks rate 43).
    assert_eq!(level_pre_split(CompressionLevel::Default), Some(1));
    // Better is a pure alias for level 7 (lazy): same as Level(7).
    assert_eq!(
        level_pre_split(CompressionLevel::Better),
        level_pre_split(CompressionLevel::Level(7)),
    );
    // Best resolves to the level-13 table row (btlazy2): pin it to that
    // numeric route so the named path can't drift from the pre-split
    // table.
    assert_eq!(
        level_pre_split(CompressionLevel::Best),
        level_pre_split(CompressionLevel::Level(13)),
    );
    assert_eq!(level_pre_split(CompressionLevel::Level(2)), Some(0)); // fast
    assert_eq!(level_pre_split(CompressionLevel::Level(4)), Some(1)); // dfast
    assert_eq!(level_pre_split(CompressionLevel::Level(5)), Some(2)); // greedy
    assert_eq!(level_pre_split(CompressionLevel::Level(7)), Some(2)); // lazy (depth 1)
    assert_eq!(level_pre_split(CompressionLevel::Level(8)), Some(1)); // lazy2 lower bound
    assert_eq!(level_pre_split(CompressionLevel::Level(11)), Some(1)); // lazy2 (depth 2)
    assert_eq!(level_pre_split(CompressionLevel::Level(12)), Some(1)); // lazy2 upper bound
    assert_eq!(level_pre_split(CompressionLevel::Level(13)), Some(1)); // btlazy2 lower bound
    assert_eq!(level_pre_split(CompressionLevel::Level(15)), Some(1)); // btlazy2 (depth 2)
    assert_eq!(level_pre_split(CompressionLevel::Level(16)), Some(2)); // btopt
    assert_eq!(level_pre_split(CompressionLevel::Level(22)), Some(2)); // btultra2
}

/// A homogeneous but periodic multi-block stream at the lazy (L7), lazy2
/// (L8) and btlazy2 (L15) levels round-trips and stays small: the lazy2 /
/// btlazy2 tier (byChunks rate 43, two steps coarser than upstream's rate 5,
/// see `pre_split_for`) must not over-split the periodic input the way
/// upstream does (528 bytes on 512 KB vs 189 for the lazy tier). The size
/// bound against the reference is asserted in `ffi-bench`
/// (`periodic_stream_presplit_matches_reference`).
#[test]
fn periodic_stream_roundtrips_at_every_presplit_tier() {
    use crate::encoding::{CompressionLevel, compress_slice_to_vec};
    const LINES: &[&str] = &[
        "ts=2026-03-26T21:39:28Z level=INFO msg=\"flush memtable\" tenant=demo table=orders region=eu-west\n",
        "ts=2026-03-26T21:39:29Z level=INFO msg=\"rotate segment\" tenant=demo table=orders region=eu-west\n",
        "ts=2026-03-26T21:39:30Z level=INFO msg=\"compact level\" tenant=demo table=orders region=eu-west\n",
        "ts=2026-03-26T21:39:31Z level=INFO msg=\"write block\" tenant=demo table=orders region=eu-west\n",
    ];
    // 512 KB = 4 upstream zstd blocks, enough for the cascade to manifest.
    let target = 512 * 1024usize;
    let mut data = Vec::with_capacity(target);
    let mut i = 0;
    while data.len() < target {
        let line = LINES[i % LINES.len()].as_bytes();
        let take = line.len().min(target - data.len());
        data.extend_from_slice(&line[..take]);
        i += 1;
    }
    let l7 = compress_slice_to_vec(&data, CompressionLevel::Level(7)); // lazy depth1
    let l8 = compress_slice_to_vec(&data, CompressionLevel::Level(8)); // lazy2
    let l15 = compress_slice_to_vec(&data, CompressionLevel::Level(15)); // btlazy2
    // 512 KB of four repeating lines: every tier stays within a few block
    // headers of the ~190-byte single-block frame (upstream: 189 / 528 / 528).
    for (level, out) in [(7, &l7), (8, &l8), (15, &l15)] {
        assert!(
            out.len() < 1024,
            "L{level} periodic stream ballooned: {} bytes",
            out.len()
        );
    }
    for out in [&l7, &l8, &l15] {
        let mut decoder = FrameDecoder::new();
        let mut round = Vec::with_capacity(data.len());
        decoder
            .decode_all_to_vec(out, &mut round)
            .expect("decode periodic stream");
        assert_eq!(round, data, "periodic stream roundtrip mismatch");
    }
}

/// End-to-end: a 256 KB payload whose SECOND 128 KB upstream zstd block carries
/// an intra-block fingerprint transition, compressed at Level(5)
/// (greedy, the pre-split path this revision routes through the cheap
/// chunk splitter), round-trips through the crate's own decoder.
///
/// The transition lives in the second block on purpose: the upstream zstd
/// `savings < 3` gate skips splitting the first block (savings start at
/// 0), so the first block is a homogeneous compressible run that banks
/// savings, and the second block is the one whose intra-block transition
/// `split_block_by_chunks()` resolves into a sub-block boundary (the
/// `pending_input.split_off(...)` path). The test asserts that split
/// decision directly so it cannot silently stop exercising the path if
/// the fixture or params drift, then proves the emitted split frame
/// round-trips. Level 13 (lazy) no longer pre-splits, hence Level 5.
#[test]
fn greedy_chunk_split_roundtrips_through_own_decoder() {
    use crate::encoding::CompressionLevel;
    let mut data = vec![0u8; 256 * 1024];
    // First 128 KB: homogeneous low-entropy run (compressible, banks
    // the savings the upstream zstd gate needs). Second 128 KB: low-entropy run
    // for its first half, then a counter sequence: a clear intra-block
    // fingerprint transition at the 192 KB midpoint for the chunk
    // splitter to find.
    for (i, byte) in data.iter_mut().enumerate() {
        *byte = if i < 192 * 1024 {
            (i & 0x07) as u8
        } else {
            (i % 251 + 1) as u8
        };
    }

    // Directly assert the chunk splitter resolves the second block's
    // intra-block transition into a sub-block boundary once savings have
    // accrued (the compressible first block banks well over the gate).
    let second_block = &data[128 * 1024..];
    let split = super::optimal_block_size(
        CompressionLevel::Level(5),
        second_block,
        second_block.len(),
        MAX_BLOCK_SIZE as usize,
        100,
    );
    assert!(
        split < MAX_BLOCK_SIZE as usize,
        "second upstream zstd block must chunk-split at its intra-block transition, got {split}",
    );

    let mut compressed = Vec::new();
    let mut compressor = FrameCompressor::new(CompressionLevel::Level(5));
    compressor.set_source(data.as_slice());
    compressor.set_drain(&mut compressed);
    compressor.compress();

    let mut decoder = FrameDecoder::new();
    let mut source = compressed.as_slice();
    decoder
        .reset(&mut source)
        .expect("frame header should parse");
    while !decoder.is_finished() {
        decoder
            .decode_blocks(&mut source, crate::decoding::BlockDecodingStrategy::All)
            .expect("decode should succeed");
    }
    let mut decoded = Vec::with_capacity(data.len());
    decoder.collect_to_writer(&mut decoded).unwrap();
    assert_eq!(decoded, data, "roundtrip must reproduce the input verbatim");
}

/// Outside-diff coverage for the FAST one-shot path.
/// `compress_slice_to_vec` / `compress_independent_frame` on a Fast level
/// routes through `run_borrowed_block_loop` (not the owned loop the test
/// above covers), which must honour `optimal_block_size` and emit a
/// sub-`MAX_BLOCK_SIZE` boundary rather than fixed 128 KiB blocks. A
/// 256 KiB input is two 128 KiB blocks when unsplit; a chunk boundary in
/// the second block yields >= 3 decoded blocks, asserted on the round-trip.
#[test]
fn fast_oneshot_borrowed_split_emits_subblock() {
    use crate::encoding::CompressionLevel;
    // First 192 KiB: homogeneous zero run (banks the savings the split
    // gate needs). The second 128 KiB block flips to a counter sequence
    // at its 64 KiB midpoint (the 192 KiB mark) — a fingerprint
    // transition the Fast from-borders splitter (split level 0) resolves
    // into a sub-block boundary.
    let mut data = vec![0u8; 256 * 1024];
    for (i, byte) in data.iter_mut().enumerate() {
        if i >= 192 * 1024 {
            *byte = (i % 251 + 1) as u8;
        }
    }

    // Pin the splitter decision for the Fast path directly (mirrors the
    // greedy test): the second upstream zstd block must resolve to a sub-block
    // boundary, so the >= 3 block count below cannot pass vacuously.
    let second_block = &data[128 * 1024..];
    assert!(
        super::optimal_block_size(
            CompressionLevel::Fastest,
            second_block,
            second_block.len(),
            MAX_BLOCK_SIZE as usize,
            100,
        ) < MAX_BLOCK_SIZE as usize,
        "fixture must resolve to a sub-block split in the second upstream zstd block",
    );

    // Drive the borrowed one-shot route explicitly (Fast level ->
    // run_borrowed_block_loop via compress_independent_frame).
    let mut compressor: FrameCompressor = FrameCompressor::new(CompressionLevel::Fastest);
    let frame = compressor.compress_independent_frame(&data);

    let mut decoder = FrameDecoder::new();
    let mut source = frame.as_slice();
    decoder
        .reset(&mut source)
        .expect("frame header should parse");
    while !decoder.is_finished() {
        decoder
            .decode_blocks(&mut source, crate::decoding::BlockDecodingStrategy::All)
            .expect("decode should succeed");
    }
    let mut decoded = Vec::with_capacity(data.len());
    decoder.collect_to_writer(&mut decoded).unwrap();
    assert_eq!(decoded, data, "roundtrip must reproduce the input verbatim");
    assert!(
        decoder.blocks_decoded() >= 3,
        "fast one-shot borrowed path must split the second upstream zstd block \
             (256 KiB unsplit = 2 blocks), got {} blocks",
        decoder.blocks_decoded(),
    );
}

/// Regression: `set_parameters` must key `literal_compression_disabled` off the
/// RESOLVED strategy / target length (C `ZSTD_literalsCompressionIsDisabled` =
/// `strategy == fast && targetLength > 0`), not the signed level. A negative
/// level overridden onto a non-fast strategy must keep literal (Huffman)
/// compression enabled.
#[cfg(feature = "std")]
#[test]
fn set_parameters_keeps_literals_compressed_under_nonfast_strategy_override() {
    use super::CompressionLevel;
    use crate::encoding::parameters::{CompressionParameters, Strategy};
    let data = vec![0xABu8; 256];
    let mut out = Vec::new();
    let mut compressor = FrameCompressor::new(CompressionLevel::Level(3));
    compressor.set_source(data.as_slice());
    compressor.set_drain(&mut out);
    let params = CompressionParameters::builder(CompressionLevel::Level(-5))
        .strategy(Strategy::Btultra2)
        .build()
        .expect("valid params");
    compressor.set_parameters(&params);
    assert!(
        !compressor.state.literal_compression_disabled,
        "a non-fast strategy override on a negative level must keep literals compressed",
    );
}

/// Regression: `set_compression_level` must resync
/// `state.literal_compression_disabled` so reusing a compressor and switching to
/// a negative level emits raw literals (matching C
/// `ZSTD_literalsCompressionIsDisabled`), not the Huffman-compressed literals
/// carried over from the prior non-negative level.
#[cfg(feature = "std")]
#[test]
fn set_compression_level_resyncs_literal_disable_for_negatives() {
    use super::CompressionLevel;
    let data = vec![0xABu8; 256];
    let mut out = Vec::new();
    // Construction at a non-negative level keeps literal (Huffman) compression on.
    let mut compressor = FrameCompressor::new(CompressionLevel::Level(3));
    compressor.set_source(data.as_slice());
    compressor.set_drain(&mut out);
    assert!(
        !compressor.state.literal_compression_disabled,
        "L3 construction must leave literal compression enabled",
    );
    // Switching to a negative level must immediately disable it.
    compressor.set_compression_level(CompressionLevel::Level(-5));
    assert!(
        compressor.state.literal_compression_disabled,
        "set_compression_level to a negative level must disable literal compression",
    );
}

/// Regression: `set_compression_level` must forget a literal compression mode
/// installed by `set_parameters` along with the other overrides, so a
/// compressor switched back to a bare level emits the frame that level emits
/// on its own, not one carrying the previous parameters' raw literals.
#[cfg(feature = "std")]
#[test]
fn set_compression_level_forgets_the_literal_compression_mode() {
    use super::CompressionLevel;
    use crate::encoding::{CompressionParameters, LiteralCompressionMode};

    // Literal-heavy input: 32 symbols with nothing for the match finder, so
    // whether the literals are Huffman-coded decides the frame size.
    let text: Vec<u8> = (0..8192u32)
        .map(|i| b'a' + (i.wrapping_mul(2_654_435_761) >> 27) as u8)
        .collect();
    let level = CompressionLevel::Level(3);
    let raw_literals = CompressionParameters::builder(level)
        .literal_compression(LiteralCompressionMode::Disable)
        .build()
        .unwrap();

    let mut reused: FrameCompressor = FrameCompressor::new(level);
    reused.set_parameters(&raw_literals);
    let with_raw = reused.compress_independent_frame(&text);
    reused.set_compression_level(level);
    let after_switch = reused.compress_independent_frame(&text);

    let mut fresh: FrameCompressor = FrameCompressor::new(level);
    let plain = fresh.compress_independent_frame(&text);
    assert!(
        with_raw.len() > plain.len(),
        "the fixture must make the mode visible: {} vs {} bytes",
        with_raw.len(),
        plain.len()
    );
    assert_eq!(
        after_switch, plain,
        "a bare level after set_parameters must compress as that level alone does"
    );
}

/// Regression: `set_compression_level` followed by `compress()` must
/// refresh `state.strategy_tag` through the reset-time sync so the
/// literal-compression gates (`min_literals_to_compress`,
/// `min_gain`) use the NEW level's strategy. Picks a level pair
/// that genuinely crosses strategy bands — `Fastest` resolves to
/// `Fast`, `Level(20)` resolves to `BtUltra2` — so a missed sync
/// would leave the construction-time tag visible and trip the
/// assertion. `CompressionLevel::Best` would also pass type-wise
/// but resolves to `Lazy` today, which keeps `min_literals_to_compress`
/// in the same `shift=3 → 64-byte` band as `Fast` and weakens the
/// signal that the gate floor actually moved.
#[cfg(feature = "std")]
#[test]
fn set_compression_level_then_compress_refreshes_strategy_tag() {
    use super::CompressionLevel;
    use crate::encoding::strategy::StrategyTag;

    let data = vec![0xABu8; 256];
    let mut out = Vec::new();
    let mut compressor = FrameCompressor::new(CompressionLevel::Fastest);
    let initial_tag = compressor.state.strategy_tag;
    assert_eq!(
        initial_tag,
        StrategyTag::for_compression_level(CompressionLevel::Fastest),
        "construction-time strategy_tag must reflect initial level",
    );

    // Switch to a level whose resolved strategy lives in a different
    // band, then run a full compress cycle — the matcher.reset()
    // inside `compress` is the only site that can refresh the tag.
    let new_level = CompressionLevel::Level(20);
    compressor.set_compression_level(new_level);
    compressor.set_source(data.as_slice());
    compressor.set_drain(&mut out);
    compressor.compress();

    let new_tag = compressor.state.strategy_tag;
    let expected = StrategyTag::for_compression_level(new_level);
    assert_eq!(
        new_tag, expected,
        "strategy_tag must follow set_compression_level → compress, \
             got {new_tag:?} expected {expected:?}",
    );
    assert_eq!(
        expected,
        StrategyTag::BtUltra2,
        "test fixture invariant: Level(20) must resolve to BtUltra2 \
             so the post-switch tag visibly crosses the band boundary",
    );
    assert_ne!(
        new_tag, initial_tag,
        "test fixture invariant: chosen levels must resolve to \
             different StrategyTag variants",
    );
}

/// Magicless mode (`ZSTD_f_zstd1_magicless`): encoded frame
/// MUST NOT start with the 4-byte magic prefix, AND must
/// round-trip through a magicless-aware decoder.
#[test]
fn magicless_frame_omits_magic_and_roundtrips() {
    use crate::common::MAGIC_NUM;
    let input: alloc::vec::Vec<u8> = (0..512u32).map(|i| (i ^ 0xA5) as u8).collect();

    // Encode with magicless = true.
    let mut output: Vec<u8> = Vec::new();
    let mut compressor = FrameCompressor::new(super::CompressionLevel::Default);
    compressor.set_magicless(true);
    compressor.set_source(input.as_slice());
    compressor.set_drain(&mut output);
    compressor.compress();

    // 1. Encoded output must NOT begin with the zstd magic number.
    assert!(
        !output.starts_with(&MAGIC_NUM.to_le_bytes()),
        "magicless frame must omit the 4-byte magic prefix",
    );

    // 2. A magicless-aware decoder must round-trip the payload.
    let mut decoder = crate::decoding::FrameDecoder::new();
    decoder.set_magicless(true);
    let mut cursor: &[u8] = output.as_slice();
    decoder.init(&mut cursor).expect("magicless init");
    decoder
        .decode_blocks(&mut cursor, crate::decoding::BlockDecodingStrategy::All)
        .expect("decode_blocks");
    let mut decoded: Vec<u8> = Vec::new();
    decoder
        .collect_to_writer(&mut decoded)
        .expect("collect_to_writer");
    assert_eq!(decoded, input, "magicless roundtrip must preserve bytes");

    // 3. A standard (magicful) decoder MUST reject a magicless
    //    frame at the header-read step — the first 4 bytes are
    //    the frame-header descriptor + window / dictionary / FCS
    //    metadata, not the magic. We accept either
    //    `BadMagicNumber` (typical case: first 4 bytes don't
    //    match `MAGIC_NUM` and don't fall in the skippable-frame
    //    magic range) or `SkipFrame` (rare: the first 4 bytes
    //    coincidentally land in `0x184D2A50..=0x184D2A5F`). Both
    //    prove the standard decoder did not treat the bytes as a
    //    real magicful frame.
    use crate::decoding::errors::{FrameDecoderError, ReadFrameHeaderError};
    let mut std_decoder = crate::decoding::FrameDecoder::new();
    let std_init = std_decoder.init(output.as_slice());
    match std_init {
        Err(FrameDecoderError::ReadFrameHeaderError(
            ReadFrameHeaderError::BadMagicNumber(_) | ReadFrameHeaderError::SkipFrame { .. },
        )) => {}
        other => panic!(
            "standard decoder must reject a magicless frame with \
                 ReadFrameHeaderError::BadMagicNumber or SkipFrame, got {other:?}",
        ),
    }
}

/// A reused `FrameCompressor` must emit byte-identical frames to a
/// fresh compressor per input across both the borrowed (Fast) and
/// owned (Dfast/Lazy/Greedy/Uncompressed) backends. This proves
/// `prepare_frame` fully resets the per-frame state (matcher window,
/// content hasher, FSE/Huffman seeds) between independent frames; a
/// missed reset would corrupt frame N>=2's header checksum or matches.
/// Each emitted frame must also round-trip.
#[test]
fn compress_independent_frame_reuse_matches_fresh_and_roundtrips() {
    use crate::encoding::{CompressionLevel, compress_slice_to_vec};
    let levels = [
        CompressionLevel::Uncompressed,
        CompressionLevel::Fastest,
        CompressionLevel::Default,
        CompressionLevel::Better,
        CompressionLevel::Best,
        CompressionLevel::Level(5),
    ];
    let inputs: Vec<Vec<u8>> = vec![
        Vec::new(),
        vec![0x00],
        b"the quick brown fox jumps over the lazy dog\n".to_vec(),
        vec![0x7Eu8; 50_000],          // highly compressible
        generate_data(0xABCD, 70_000), // pseudo-random
        generate_data(0x1234, 200_000),
    ];
    for level in levels {
        let mut cctx: FrameCompressor = FrameCompressor::new(level);
        for data in &inputs {
            let reused = cctx.compress_independent_frame(data);
            let fresh = compress_slice_to_vec(data, level);
            assert_eq!(
                reused,
                fresh,
                "reused frame != fresh frame for len={} level={:?}",
                data.len(),
                level,
            );
            let mut decoder = FrameDecoder::new();
            let mut decoded = Vec::with_capacity(data.len());
            decoder.decode_all_to_vec(&reused, &mut decoded).unwrap();
            assert_eq!(
                decoded,
                *data,
                "roundtrip failed for len={} level={:?}",
                data.len(),
                level,
            );
        }
    }
}

/// The same promise across the lazy and optimal bands, where the match
/// finder carries the most state between blocks: a compressor reused frame
/// after frame writes exactly what a fresh one writes for each input, so no
/// frame's output depends on what came before it.
#[test]
fn compress_independent_frame_reuse_matches_fresh_on_the_optimal_band() {
    use crate::encoding::{CompressionLevel, compress_slice_to_vec};
    let text: Vec<u8> = (0..3_000u32)
        .flat_map(|i| alloc::format!("row {} key {} val {}\n", i % 97, i % 13, i % 7).into_bytes())
        .collect();
    let inputs: Vec<Vec<u8>> = vec![
        text[..5_000].to_vec(),
        generate_data(0xABCD, 20_000),
        text.clone(),
        generate_data(0x1234, 9_000),
        text[1_000..1_700].to_vec(),
    ];
    let mut diverged = Vec::new();
    for level in [12, 16, 17, 19, 22] {
        let level = CompressionLevel::Level(level);
        let mut cctx: FrameCompressor = FrameCompressor::new(level);
        for (index, data) in inputs.iter().enumerate() {
            let reused = cctx.compress_independent_frame(data);
            let fresh = compress_slice_to_vec(data, level);
            if reused != fresh {
                diverged.push(alloc::format!(
                    "{level:?} input {index}: {} bytes reused against {} fresh",
                    reused.len(),
                    fresh.len()
                ));
            }
        }
    }
    assert!(diverged.is_empty(), "{diverged:#?}");
}

/// A compressor kept for one-shot frame after frame, with its parameters set
/// again before each (what a C context compressing through
/// `ZSTD_compress2` does), writes each 4 KiB piece of a stream exactly as a
/// fresh compressor with the same parameters writes it, at every level.
#[test]
fn a_kept_compressor_writes_each_small_piece_as_a_fresh_one() {
    use crate::encoding::{CompressionLevel, CompressionParameters};
    let text: Vec<u8> = (0..12_000u32)
        .flat_map(|i| {
            alloc::format!(
                "ts={} host=h{} level={} msg=event {} took {}ms\n",
                1_700_000_000 + i * 7,
                i % 13,
                ["info", "warn", "debug"][(i % 3) as usize],
                i % 211,
                (i * 37) % 997
            )
            .into_bytes()
        })
        .collect();
    let mut diverged = Vec::new();
    for level in [-3, 1, 2, 3, 4, 5, 6, 7, 9, 12, 13, 16, 19, 22] {
        let level = CompressionLevel::from_level(level);
        let params = CompressionParameters::builder(level).build().unwrap();
        let mut kept: FrameCompressor = FrameCompressor::new(level);
        for (index, piece) in text.chunks(4096).take(40).enumerate() {
            kept.set_parameters(&params);
            let reused = kept.compress_independent_frame(piece);
            let mut fresh: FrameCompressor = FrameCompressor::new(level);
            fresh.set_parameters(&params);
            let expected = fresh.compress_independent_frame(piece);
            if reused != expected {
                let mut decoded = Vec::with_capacity(piece.len());
                let decodes = FrameDecoder::new()
                    .decode_all_to_vec(&reused, &mut decoded)
                    .is_ok()
                    && decoded == piece;
                diverged.push(alloc::format!(
                    "{level:?} piece {index}: {} bytes reused against {} fresh, decodes: {decodes}",
                    reused.len(),
                    expected.len()
                ));
            }
        }
    }
    assert!(diverged.is_empty(), "{diverged:#?}");
}

/// `compress_independent_frame_into` must replace (not append to) the
/// caller's buffer each call, so a smaller frame after a larger one
/// yields exactly the smaller frame, and the reused buffer's content
/// matches a fresh compression of the same input.
#[test]
fn compress_independent_frame_into_replaces_buffer_contents() {
    use crate::encoding::{CompressionLevel, compress_slice_to_vec};
    let large = vec![0x11u8; 40_000];
    let small = b"short payload".to_vec();
    let mut cctx: FrameCompressor = FrameCompressor::new(CompressionLevel::Default);
    let mut out = Vec::new();
    cctx.compress_independent_frame_into(&large, &mut out);
    let frame_large = out.clone();
    // Reusing the same buffer for a smaller frame must clear it first.
    cctx.compress_independent_frame_into(&small, &mut out);
    assert_eq!(
        out,
        compress_slice_to_vec(&small, CompressionLevel::Default),
        "reused buffer must hold exactly the second frame",
    );
    // The first frame, captured before reuse, still round-trips.
    let mut decoder = FrameDecoder::new();
    let mut decoded = Vec::with_capacity(large.len());
    decoder
        .decode_all_to_vec(&frame_large, &mut decoded)
        .unwrap();
    assert_eq!(decoded, large);
}

/// A sticky dictionary set once on a reused compressor must be primed
/// into every independent frame (mirroring `ZSTD_CCtx_loadDictionary`):
/// each frame decodes with the dictionary and is byte-identical to a
/// fresh compressor carrying the same dictionary. This proves
/// `prepare_frame` re-primes the dictionary (matcher content + offset
/// history + entropy seed) every call rather than only on the first.
#[test]
fn compress_independent_frame_reuses_sticky_dictionary() {
    use crate::encoding::CompressionLevel;
    let dict_raw = include_bytes!("../../../dict_tests/dictionary");
    let dict_content = crate::decoding::Dictionary::decode_dict(dict_raw).unwrap();
    let mut payload_a = Vec::new();
    for _ in 0..8 {
        payload_a.extend_from_slice(&dict_content.dict_content[..2048]);
    }
    let payload_b = b"a different second frame payload, still dict-attached".to_vec();
    let inputs = [payload_a, payload_b];

    let mut cctx: FrameCompressor = FrameCompressor::new(CompressionLevel::Fastest);
    cctx.set_dictionary_from_bytes(dict_raw)
        .expect("dictionary bytes should parse");

    for data in &inputs {
        let reused = cctx.compress_independent_frame(data);
        // Fresh compressor carrying the same sticky dictionary.
        let mut fresh_enc: FrameCompressor = FrameCompressor::new(CompressionLevel::Fastest);
        fresh_enc
            .set_dictionary_from_bytes(dict_raw)
            .expect("dictionary bytes should parse");
        let fresh = fresh_enc.compress_independent_frame(data);
        assert_eq!(
            reused,
            fresh,
            "reused dict frame != fresh dict frame, len={}",
            data.len(),
        );
        // Round-trip with the dictionary on the decode side.
        let dict_for_decoder = crate::decoding::Dictionary::decode_dict(dict_raw).unwrap();
        let mut decoder = FrameDecoder::new();
        decoder.add_dict(dict_for_decoder).unwrap();
        let mut decoded = Vec::with_capacity(data.len());
        decoder.decode_all_to_vec(&reused, &mut decoded).unwrap();
        assert_eq!(&decoded, data, "dict roundtrip failed, len={}", data.len());
    }
}

/// Walk a frame's block list, returning `(block_type, block_size, last)` per
/// physical block. `block_type`: 0 = Raw, 1 = RLE, 2 = Compressed.
fn frame_block_list(frame: &[u8]) -> Vec<(u8, usize, bool)> {
    let desc = frame[4];
    let fcs_flag = desc >> 6;
    let single_segment = (desc >> 5) & 1 == 1;
    let checksum = (desc >> 2) & 1 == 1;
    let dict_id_bytes = match desc & 3 {
        0 => 0,
        1 => 1,
        2 => 2,
        _ => 4,
    };
    let fcs_bytes = match fcs_flag {
        0 => usize::from(single_segment),
        1 => 2,
        2 => 4,
        _ => 8,
    };
    let mut pos = 4 + 1 + usize::from(!single_segment) + dict_id_bytes + fcs_bytes;
    let end = frame.len() - if checksum { 4 } else { 0 };
    let mut blocks = Vec::new();
    while pos + 3 <= end {
        let h =
            frame[pos] as usize | (frame[pos + 1] as usize) << 8 | (frame[pos + 2] as usize) << 16;
        let last = h & 1 == 1;
        let btype = ((h >> 1) & 3) as u8;
        let bsize = h >> 3;
        let advance = if btype == 1 { 1 } else { bsize };
        blocks.push((btype, bsize, last));
        pos += 3 + advance;
        if last {
            break;
        }
    }
    blocks
}

/// An input that is an exact multiple of `MAX_BLOCK_SIZE` must NOT emit a
/// spurious trailing empty Raw block: the last REAL block carries the
/// `last_block` flag, matching the C encoder on `ZSTD_e_end`. Exercises both
/// the one-shot slice loop (`compress_independent_frame`) and the streaming
/// `Read` loop (`set_source` + `compress`), on incompressible input (Raw
/// blocks) and highly compressible input (Compressed/RLE blocks). Before the
/// fix, each frame ended with an extra `R0!` block (3 wasted bytes).
#[test]
fn exact_block_multiple_marks_last_real_block() {
    // The fix is driven by `block_capacity`, so cover both the default 128 KiB
    // cap (`None`) and a smaller configured `target_block_size` — the EOF path
    // must mark the last real block in both.
    const CUSTOM_CAP: u32 = 16 * 1024;
    for &(cap, target) in &[
        (MAX_BLOCK_SIZE as usize, None),
        (CUSTOM_CAP as usize, Some(CUSTOM_CAP)),
    ] {
        for &nblk in &[1usize, 2, 3] {
            for &compressible in &[false, true] {
                let input: Vec<u8> = if compressible {
                    vec![0x7Au8; cap * nblk]
                } else {
                    generate_data(0xC0FF_EE11, cap * nblk)
                };

                // One-shot slice path.
                let mut oneshot: FrameCompressor =
                    FrameCompressor::new(super::CompressionLevel::Default);
                oneshot.set_target_block_size(target);
                let frame_os = oneshot.compress_independent_frame(&input);
                let blocks_os = frame_block_list(&frame_os);
                let last_os = *blocks_os.last().expect("at least one block");
                assert!(
                    last_os.2,
                    "one-shot last block must set last_block (cap={cap}, nblk={nblk}, compressible={compressible}): {blocks_os:?}"
                );
                assert!(
                    !(last_os.0 == 0 && last_os.1 == 0),
                    "one-shot must not emit a trailing empty Raw block (cap={cap}, nblk={nblk}, compressible={compressible}): {blocks_os:?}"
                );

                // Streaming Read loop.
                let mut output = Vec::new();
                let mut streaming = FrameCompressor::new(super::CompressionLevel::Default);
                streaming.set_target_block_size(target);
                streaming.set_source(input.as_slice());
                streaming.set_drain(&mut output);
                streaming.compress();
                let blocks_st = frame_block_list(&output);
                let last_st = *blocks_st.last().expect("at least one block");
                assert!(
                    last_st.2,
                    "streaming last block must set last_block (cap={cap}, nblk={nblk}, compressible={compressible}): {blocks_st:?}"
                );
                assert!(
                    !(last_st.0 == 0 && last_st.1 == 0),
                    "streaming must not emit a trailing empty Raw block (cap={cap}, nblk={nblk}, compressible={compressible}): {blocks_st:?}"
                );

                // Both frames must round-trip back to the original bytes.
                for frame in [&frame_os, &output] {
                    let mut decoder = FrameDecoder::new();
                    let mut decoded = Vec::with_capacity(input.len());
                    decoder.decode_all_to_vec(frame, &mut decoded).unwrap();
                    assert_eq!(
                        decoded, input,
                        "roundtrip mismatch (cap={cap}, nblk={nblk}, compressible={compressible})"
                    );
                }
            }
        }
    }
}

#[test]
fn dict_compress_bt_level_tiny_source_round_trips_through_prime_dms_bt() {
    // End-to-end cover for the dictionary match binary-tree (`prime_dms_bt`, a
    // ZSTD_dictMatchState analog built only on the BT strategies, level >= 13):
    // compress a tiny source with a small raw-content dictionary at level 19
    // (BtUltra2) and round-trip it. This drives the BT dict-prime + dict-match
    // search path that non-BT levels (Fast/Dfast/Lazy) never reach.
    //
    // The `prime_dms_bt` dms-table sizing previously used
    // `ceil_log2(region).clamp(10, hash_log)`, which panicked ("min > max") when
    // the matcher's `hash_log` adjusted below the 10 floor — the exact bound
    // arithmetic is pinned directly by `storage::dms_hash_log_tests`. On this
    // build the window-log floor keeps `hash_log >= 10` so this end-to-end path
    // stays above the boundary; the unit test exercises `hash_log < 10`.
    let raw_dict: Vec<u8> = (0..100u32)
        .map(|i| (i.wrapping_mul(2_654_435_761) >> 24) as u8)
        .collect();
    let dict_id = 1u32;
    let dict_for_encoder =
        crate::decoding::Dictionary::from_raw_content(dict_id, raw_dict.clone()).unwrap();
    let dict_for_decoder =
        crate::decoding::Dictionary::from_raw_content(dict_id, raw_dict).unwrap();

    let data = b"hello world".to_vec();

    let mut compressor: FrameCompressor =
        FrameCompressor::new(super::CompressionLevel::from_level(19));
    compressor
        .set_dictionary(dict_for_encoder)
        .expect("raw-content dictionary should attach");
    // Runs the BT dict-prime + dict-match path end to end.
    let out = compressor.compress_independent_frame(data.as_slice());

    let mut decoder = FrameDecoder::new();
    decoder.add_dict(dict_for_decoder).unwrap();
    let mut decoded = Vec::with_capacity(data.len());
    decoder
        .decode_all_to_vec(&out, &mut decoded)
        .expect("dict BT-level frame should round-trip");
    assert_eq!(decoded, data);
}

/// Pseudo-random bytes (no self-repeats), so a match into them can only
/// come from the dictionary or a verbatim copy.
fn noise_bytes(len: usize, seed: u32) -> Vec<u8> {
    let mut x = seed;
    (0..len)
        .map(|_| {
            x = x.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
            (x >> 24) as u8
        })
        .collect()
}

/// Noise the incompressibility classifier actually calls incompressible.
///
/// A test that needs the raw-skip to fire cannot use [`noise_bytes`]: a
/// truncated linear congruential sequence leaves enough structure in the low
/// bits that the classifier's repeat-bucket test rejects it, so the skip never
/// engages and the test passes whether or not the thing it guards works. This
/// runs the counter through a 64-bit mixer, which the classifier reads as
/// random — the same verdict it gives real entropy.
fn indistinguishable_noise_bytes(len: usize, seed: u64) -> Vec<u8> {
    (0..len as u64)
        .map(|i| {
            let mut x = i.wrapping_add(seed).wrapping_mul(0x9E37_79B9_7F4A_7C15);
            x ^= x >> 30;
            x = x.wrapping_mul(0xBF58_476D_1CE4_E5B9);
            x ^= x >> 27;
            (x >> 32) as u8
        })
        .collect()
}

/// The other half of removing the pre-search skip: input that really is
/// incompressible must still come out at its own size.
///
/// Dropping the skip means this payload is now searched in full and found to
/// have nothing, so it reaches a raw block through the ordinary
/// compressed-is-not-smaller fallback instead of being written off up front.
/// That fallback is the entire reason the skip was safe to delete, and nothing
/// pinned it: the repeat test above would still pass if raw emission broke and
/// random input started expanding. Overhead here is frame and block framing
/// only, tens of bytes on a megabyte.
#[test]
fn genuinely_random_input_still_comes_out_at_its_own_size() {
    let payload = noise_bytes(1024 * 1024, 0x5EED);

    for level in [
        super::CompressionLevel::Level(1),
        super::CompressionLevel::Level(3),
        super::CompressionLevel::Level(5),
        super::CompressionLevel::Level(9),
    ] {
        let mut compressor: FrameCompressor = FrameCompressor::new(level);
        let frame = compressor.compress_independent_frame(&payload);
        assert!(
            frame.len() <= payload.len() + 256,
            "{level:?}: incompressible input must not expand beyond framing \
             ({} of {})",
            frame.len(),
            payload.len(),
        );
        let mut decoder = FrameDecoder::new();
        let mut decoded = Vec::with_capacity(payload.len());
        decoder
            .decode_all_to_vec(&frame, &mut decoded)
            .expect("frame decodes");
        assert_eq!(decoded, payload, "{level:?}: round trip");
    }
}

/// Regression: high-entropy input that REPEATS must be searched, not written
/// off by its own appearance.
///
/// The encoder samples a block for entropy and, when the sample looks random,
/// may emit it raw without searching. Both halves of this payload look random
/// to any such sample, so a check that reads only the block's own bytes never
/// compares the second half against the first and throws a 256 KiB match away:
/// the frame comes out the size of its input. What keeps the skip honest is
/// the probe of the match table that runs beside it, which sees the first half
/// already indexed.
///
/// The payload has to be noise the classifier actually calls random, or the
/// skip never engages and this passes without exercising anything.
#[test]
fn repeated_high_entropy_input_is_searched_not_written_off() {
    let half = indistinguishable_noise_bytes(256 * 1024, 0xBEEF);
    let mut payload = half.clone();
    payload.extend_from_slice(&half);

    for level in [
        super::CompressionLevel::Level(1),
        super::CompressionLevel::Level(3),
        super::CompressionLevel::Level(5),
        super::CompressionLevel::Level(9),
    ] {
        let mut compressor: FrameCompressor = FrameCompressor::new(level);
        let frame = compressor.compress_independent_frame(&payload);
        assert!(
            frame.len() * 3 < payload.len() * 2,
            "{level:?}: the second half repeats the first, so the frame must be far \
             below two thirds of the input ({} of {})",
            frame.len(),
            payload.len(),
        );
        let mut decoder = FrameDecoder::new();
        let mut decoded = Vec::with_capacity(payload.len());
        decoder
            .decode_all_to_vec(&frame, &mut decoded)
            .expect("frame decodes");
        assert_eq!(decoded, payload, "{level:?}: round trip");
    }
}

/// Regression: a COPIED dictionary (source above the 32 KiB attach cutoff)
/// whose CDict cParams select the hash-chain finder (a 4 KiB dictionary
/// resolves to a window of 2^14 or less) must be indexed at ABSOLUTE
/// positions. On a reused compressor whose primed snapshot does not fit the
/// next frame (a larger source, so another window) the dictionary is
/// re-indexed with the coordinate floor past the previous frame; relative
/// indices then fall below the window floor and the dictionary silently
/// stops matching. The warm frame must equal a cold compressor's and stay
/// smaller than the dictionary-less frame.
#[test]
fn reused_compressor_copied_chain_dictionary_frame_is_byte_identical() {
    let dict_raw = noise_bytes(4096, 7);
    // Noise with dictionary slices sprinkled in: only the dictionary can
    // supply those matches.
    let make_payload = |target: usize, seed: u32| {
        let mut payload = Vec::with_capacity(target);
        let mut i = 0usize;
        while payload.len() < target {
            payload.extend_from_slice(&noise_bytes(200, seed + i as u32));
            let at = (i * 613) % (dict_raw.len() - 64);
            payload.extend_from_slice(&dict_raw[at..at + 64]);
            i += 1;
        }
        payload
    };
    let first = make_payload(100 * 1024, 100);
    let second = make_payload(200 * 1024, 5000);
    let dict_id = 0xD1C7_0011;
    let dict = || {
        crate::decoding::Dictionary::from_raw_content(dict_id, dict_raw.clone())
            .expect("raw-content dictionary")
    };
    let level = super::CompressionLevel::Level(6);
    let mut warm: FrameCompressor = FrameCompressor::new(level);
    warm.set_dictionary(dict()).expect("dict attach");
    let _ = warm.compress_independent_frame(&first);
    let warm_frame = warm.compress_independent_frame(&second);
    let mut cold: FrameCompressor = FrameCompressor::new(level);
    cold.set_dictionary(dict()).expect("dict attach");
    let cold_frame = cold.compress_independent_frame(&second);
    assert_eq!(
        warm_frame, cold_frame,
        "a reused compressor must index the copied dictionary at absolute positions"
    );
    let mut plain: FrameCompressor = FrameCompressor::new(level);
    let no_dict = plain.compress_independent_frame(&second);
    assert!(
        cold_frame.len() < no_dict.len(),
        "the copied dictionary must supply matches ({} vs {} without it)",
        cold_frame.len(),
        no_dict.len()
    );
    for frame in [&warm_frame, &cold_frame] {
        let mut decoder = FrameDecoder::new();
        decoder.add_dict(dict()).unwrap();
        let mut decoded = Vec::with_capacity(second.len());
        decoder.decode_all_to_vec(frame, &mut decoded).unwrap();
        assert_eq!(decoded, second);
    }
}

/// Regression: a small dictionary on a btlazy2 level resolves to an optimal
/// CDict strategy (the <= 16 KiB CDict tier puts L13 at btopt), and upstream
/// runs the CDict's strategy for the frame. The frame must take that route
/// and index the dictionary into the finder it actually searches; primed
/// into tables the lazy tree never reads, the dictionary contributed
/// nothing and the frame came out as large as without it.
#[test]
fn small_dictionary_on_btlazy2_level_contributes_matches() {
    let dict_raw = noise_bytes(4096, 11);
    let mut payload = Vec::with_capacity(100 * 1024);
    let mut i = 0usize;
    while payload.len() < 100 * 1024 {
        payload.extend_from_slice(&noise_bytes(200, 300 + i as u32));
        let at = (i * 613) % (dict_raw.len() - 64);
        payload.extend_from_slice(&dict_raw[at..at + 64]);
        i += 1;
    }
    let dict_id = 0xD1C7_0012;
    let dict = || {
        crate::decoding::Dictionary::from_raw_content(dict_id, dict_raw.clone())
            .expect("raw-content dictionary")
    };
    let level = super::CompressionLevel::Level(13);
    let mut with_dict: FrameCompressor = FrameCompressor::new(level);
    with_dict.set_dictionary(dict()).expect("dict attach");
    let frame = with_dict.compress_independent_frame(&payload);
    let mut plain: FrameCompressor = FrameCompressor::new(level);
    let no_dict = plain.compress_independent_frame(&payload);
    // Every 264-byte record carries a 64-byte dictionary slice, so the
    // dictionary must take a visible bite out of the output.
    //
    // The margin is measured against a baseline that finds what it can on its
    // own. The records are drawn from a 4 KiB dictionary, so those slices also
    // recur WITHIN the payload, and the no-dict arm finds them: it went from
    // 102,444 bytes to 82,581 when the encoder stopped skipping the search on
    // blocks that merely look high-entropy. The dictionary arm did not move at
    // all across that change (79,121 either way), so the smaller margin here
    // reports a stronger baseline, not a weaker dictionary. The regression this
    // guards is the dictionary contributing NOTHING — primed into tables the
    // search never reads — which would put the two arms level.
    assert!(
        frame.len() * 100 < no_dict.len() * 97,
        "the dictionary must supply matches at L13 ({} vs {} without it)",
        frame.len(),
        no_dict.len()
    );
    let mut decoder = FrameDecoder::new();
    decoder.add_dict(dict()).unwrap();
    let mut decoded = Vec::with_capacity(payload.len());
    decoder.decode_all_to_vec(&frame, &mut decoded).unwrap();
    assert_eq!(decoded, payload);
}

/// Regression: the pre-split tier follows the EFFECTIVE strategy (upstream
/// indexes `splitLevels` by `cParams.strategy`): a strategy override on a
/// level, or a small source promoting a level, must move the tier with it.
#[test]
fn pre_split_tier_follows_the_effective_strategy() {
    use crate::encoding::{CompressionParameters, Strategy};
    let mut enc: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Level(1));
    assert_eq!(
        enc.state.pre_split,
        Some(0),
        "L1 is fast: the borders splitter"
    );
    let params = CompressionParameters::builder(super::CompressionLevel::Level(1))
        .strategy(Strategy::Btultra2)
        .build()
        .expect("valid override");
    enc.set_parameters(&params);
    assert_eq!(enc.state.pre_split, Some(2), "btultra2 override: tier 2");
    let params = CompressionParameters::builder(super::CompressionLevel::Level(1))
        .strategy(Strategy::Lazy2)
        .build()
        .expect("valid override");
    enc.set_parameters(&params);
    assert_eq!(enc.state.pre_split, Some(1), "lazy2 override: tier 1");
    let params = CompressionParameters::builder(super::CompressionLevel::Level(1))
        .strategy(Strategy::Lazy)
        .build()
        .expect("valid override");
    enc.set_parameters(&params);
    assert_eq!(enc.state.pre_split, Some(2), "lazy override: tier 2");
    // A 4 KiB source promotes L13 from btlazy2 (tier 1) to btopt (tier 2).
    let mut enc: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Level(13));
    enc.set_source_size_hint(4096);
    let params = CompressionParameters::builder(super::CompressionLevel::Level(13))
        .build()
        .expect("valid params");
    enc.set_parameters(&params);
    assert_eq!(
        enc.state.pre_split,
        Some(2),
        "L13 on a 4 KiB source is btopt"
    );
}

/// A dictionary prepared under explicit parameters runs them (upstream
/// `ZSTD_createCDict_advanced2` builds the CDict from the context's
/// parameters), so the frame state the literal gates and the block splitter
/// read records the strategy asked for, and the frame decodes. The 20 KiB
/// CDict alone would resolve L6 to lazy.
#[test]
fn dictionary_frame_runs_a_strategy_override() {
    use crate::encoding::strategy::{BackendTag, StrategyTag};
    use crate::encoding::{CompressionParameters, Strategy};
    let dict_raw = noise_bytes(20 * 1024, 5);
    let mut payload = dict_raw[4096..12 * 1024].to_vec();
    payload.extend_from_slice(&noise_bytes(8 * 1024, 9));
    let mut enc: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Level(6));
    enc.set_dictionary(
        crate::decoding::Dictionary::from_raw_content(0xD1C7_0013, dict_raw.clone()).unwrap(),
    )
    .unwrap();
    let params = CompressionParameters::builder(super::CompressionLevel::Level(6))
        .strategy(Strategy::Btultra2)
        .build()
        .expect("valid override");
    enc.set_parameters(&params);
    enc.set_source_size_hint(payload.len() as u64);
    let frame = enc.compress_independent_frame(&payload);
    assert_eq!(enc.state.strategy_tag, StrategyTag::BtUltra2);
    assert_eq!(enc.state.matcher.active_backend(), BackendTag::HashChain);
    assert_eq!(
        enc.state.pre_split,
        Some(crate::encoding::levels::config::pre_split_for(
            StrategyTag::BtUltra2,
            2
        ))
    );
    assert!(
        frame.len() < payload.len() / 2 + 64,
        "the dictionary half of the payload is found through the optimal search ({} bytes)",
        frame.len()
    );
    let mut decoder = FrameDecoder::new();
    decoder
        .add_dict(crate::decoding::Dictionary::from_raw_content(0xD1C7_0013, dict_raw).unwrap())
        .unwrap();
    let mut decoded = Vec::with_capacity(payload.len());
    decoder.decode_all_to_vec(&frame, &mut decoded).unwrap();
    assert_eq!(decoded, payload);
}

/// The bytes a set of dictionary handles alone keeps alive: one dictionary
/// held twice in the set counts once, one also held outside the set counts
/// nothing, and distinct dictionaries each count. A compressor reports the
/// dictionary it holds.
#[test]
fn exclusive_heap_size_counts_what_only_the_set_holds() {
    use crate::encoding::EncoderDictionary;
    let first = EncoderDictionary::from_serialized_or_raw_content(&noise_bytes(4096, 3)).unwrap();
    let second = EncoderDictionary::from_serialized_or_raw_content(&noise_bytes(2048, 4)).unwrap();
    let first_again = first.clone();
    assert_eq!(
        EncoderDictionary::exclusive_heap_size([&first, &first_again, &second]),
        first.heap_size() + second.heap_size()
    );
    assert_eq!(
        EncoderDictionary::exclusive_heap_size([&first, &second]),
        second.heap_size(),
        "the clone outside the set keeps the first alive on its own"
    );
    assert_eq!(
        EncoderDictionary::exclusive_heap_size(core::iter::empty::<&EncoderDictionary>()),
        0
    );

    let mut enc: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Level(3));
    assert!(enc.dictionary().is_none());
    enc.set_encoder_dictionary(second.clone())
        .expect("the dictionary attaches");
    let held = enc
        .dictionary()
        .expect("the compressor holds the dictionary");
    assert_eq!(
        EncoderDictionary::exclusive_heap_size([held, &second]),
        second.heap_size()
    );
}

/// The fast strategy hashes a key of at least 4 bytes: upstream's fast block
/// compressor takes a minMatch of 3 as 4 (zstd_fast.c,
/// `ZSTD_compressBlock_fast`: `default: /* includes case 3 */`). A min_match
/// of 3 reaches it from the knob, and from an optimal level's CDict row when
/// a dictionary frame is moved onto the fast strategy; both frames compress
/// and decode.
#[test]
fn fast_strategy_takes_a_three_byte_min_match_as_four() {
    use crate::encoding::{CompressionParameters, Strategy};
    let dict_raw = noise_bytes(16 * 1024, 5);
    let mut payload = dict_raw[2048..10 * 1024].to_vec();
    payload.extend_from_slice(&b"key=value; ".repeat(2000));
    let knob = CompressionParameters::builder(super::CompressionLevel::Level(3))
        .strategy(Strategy::Fast)
        .min_match(3)
        .build()
        .expect("valid knobs");
    let optimal_row = CompressionParameters::builder(super::CompressionLevel::Level(19))
        .strategy(Strategy::Fast)
        .build()
        .expect("valid knobs");
    for (case, params, with_dictionary) in [("knob", knob, false), ("CDict row", optimal_row, true)]
    {
        let mut enc: FrameCompressor = FrameCompressor::new(params.level());
        if with_dictionary {
            enc.set_dictionary(
                crate::decoding::Dictionary::from_raw_content(0xD1C7_001B, dict_raw.clone())
                    .unwrap(),
            )
            .unwrap();
        }
        enc.set_parameters(&params);
        let frame = enc.compress_independent_frame(&payload);
        let mut decoder = FrameDecoder::new();
        if with_dictionary {
            decoder
                .add_dict(
                    crate::decoding::Dictionary::from_raw_content(0xD1C7_001B, dict_raw.clone())
                        .unwrap(),
                )
                .unwrap();
        }
        let mut decoded = Vec::with_capacity(payload.len());
        decoder.decode_all_to_vec(&frame, &mut decoded).unwrap();
        assert!(decoded == payload, "{case}: round trip");
    }
}

/// The raw-literals gate is recomputed when the dictionary state changes
/// AFTER `set_parameters`. With a positive `target_length`, the gate turns
/// on exactly where the frame runs the fast strategy: L2 over a 200 KiB
/// source resolves to dfast, while a 4 KiB dictionary's CDict resolves it to
/// fast, so attaching the dictionary disables literal compression and
/// clearing it enables it again.
#[test]
fn literal_gate_follows_dictionary_attach_and_clear() {
    use crate::encoding::CompressionParameters;
    let dict_raw = noise_bytes(4 * 1024, 5);
    let payload = noise_bytes(200 * 1024, 9);
    let params = CompressionParameters::builder(super::CompressionLevel::Level(2))
        .target_length(8)
        .build()
        .expect("valid override");
    let mut enc: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Level(2));
    enc.set_parameters(&params);
    enc.set_source_size_hint(payload.len() as u64);
    let _ = enc.compress_independent_frame(&payload);
    assert!(
        !enc.state.literal_compression_disabled,
        "a dfast frame compresses literals whatever its targetLength"
    );
    enc.set_dictionary(
        crate::decoding::Dictionary::from_raw_content(0xD1C7_0019, dict_raw).unwrap(),
    )
    .unwrap();
    enc.set_source_size_hint(payload.len() as u64);
    let _ = enc.compress_independent_frame(&payload);
    assert!(
        enc.state.literal_compression_disabled,
        "the fast CDict with a positive targetLength leaves literals raw"
    );
    enc.clear_dictionary();
    enc.set_source_size_hint(payload.len() as u64);
    let _ = enc.compress_independent_frame(&payload);
    assert!(
        !enc.state.literal_compression_disabled,
        "without the dictionary the frame is dfast again"
    );
}

/// Regression: parameters whose level is `Uncompressed` may be set while a
/// dictionary is attached (uncompressed mode ignores the dictionary, as
/// `prepare_frame` does); the immediate strategy sync must not resolve the
/// dictionary's CDict tier for a level that has no numeric value.
#[test]
fn set_parameters_uncompressed_with_a_dictionary_attached_does_not_resolve_a_cdict() {
    use crate::encoding::CompressionParameters;
    use crate::encoding::strategy::StrategyTag;
    let dict_raw = noise_bytes(4 * 1024, 5);
    let mut enc: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Level(3));
    enc.set_dictionary(
        crate::decoding::Dictionary::from_raw_content(0xD1C7_0016, dict_raw).unwrap(),
    )
    .unwrap();
    let params = CompressionParameters::builder(super::CompressionLevel::Uncompressed)
        .build()
        .expect("valid parameters");
    enc.set_parameters(&params);
    assert_eq!(enc.state.strategy_tag, StrategyTag::Fast);
    assert_eq!(enc.state.pre_split, None);
    let payload = noise_bytes(2048, 9);
    let frame = enc.compress_independent_frame(&payload);
    let mut decoder = FrameDecoder::new();
    let mut decoded = Vec::with_capacity(payload.len());
    decoder.decode_all_to_vec(&frame, &mut decoded).unwrap();
    assert_eq!(decoded, payload);
}

/// A dictionary frame that a strategy knob moves onto the fast strategy keeps
/// its CDict row's targetLength (999 at level 22; upstream
/// `ZSTD_overrideCParams` replaces only the strategy), which is the fast
/// matcher's step, so the raw-literals gate reads that value too, as upstream
/// `ZSTD_literalsCompressionIsDisabled` reads the effective cParams.
#[test]
fn dictionary_frame_moved_onto_fast_keeps_the_cdict_target_length_in_the_literal_gate() {
    use crate::encoding::{CompressionParameters, Strategy};
    let dict_raw = noise_bytes(4 * 1024, 5);
    let params = CompressionParameters::builder(super::CompressionLevel::Level(22))
        .strategy(Strategy::Fast)
        .build()
        .expect("valid override");
    let mut enc: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Level(22));
    enc.set_dictionary(
        crate::decoding::Dictionary::from_raw_content(0xD1C7_001C, dict_raw).unwrap(),
    )
    .unwrap();
    enc.set_parameters(&params);
    assert!(
        enc.state.literal_compression_disabled,
        "after set_parameters"
    );
    let _ = enc.compress_independent_frame(&noise_bytes(2048, 9));
    assert!(
        enc.state.literal_compression_disabled,
        "at the frame's start"
    );
}

/// The raw-literals gate (`ZSTD_literalsCompressionIsDisabled`: fast
/// strategy with a positive targetLength) reads a `target_length` override on
/// a dictionary frame as on any other: the dictionary is prepared with it, so
/// the fast CDict at level 1 runs targetLength 8, not its row's 0.
#[test]
fn dictionary_frame_literal_gate_reads_a_target_length_override() {
    use crate::encoding::CompressionParameters;
    let dict_raw = noise_bytes(4 * 1024, 5);
    let mut enc: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Level(1));
    enc.set_dictionary(
        crate::decoding::Dictionary::from_raw_content(0xD1C7_0017, dict_raw).unwrap(),
    )
    .unwrap();
    let params = CompressionParameters::builder(super::CompressionLevel::Level(1))
        .target_length(8)
        .build()
        .expect("valid override");
    enc.set_parameters(&params);
    assert!(
        enc.state.literal_compression_disabled,
        "the dictionary runs targetLength 8 on the fast strategy: literals stay raw"
    );
}

/// Regression: the CDict's cParams tier is picked from the SERIALIZED
/// dictionary size (`ZSTD_createCDict(dictBuffer, dictSize, level)`), not
/// from its content length. A dictionary whose content plus the 498-byte
/// unknown-source margin fits the `<= 16 KiB` tier while its serialized
/// size does not resolves L11 to btlazy2 (the `<= 128 KiB` row), not btopt.
#[test]
fn dictionary_cdict_tier_follows_the_serialized_dictionary_size() {
    use crate::encoding::strategy::StrategyTag;
    // 16384 - 498 - 100: the content alone selects the `<= 16 KiB` tier.
    let content = noise_bytes(15_786, 11);
    let raw = serialized_dictionary(0xD1C7_0015, &content);
    assert!(content.len() + 498 <= 16 * 1024);
    assert!(raw.len() + 498 > 16 * 1024);
    let payload = noise_bytes(8 * 1024, 9);
    let mut enc: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Level(11));
    enc.set_dictionary_from_bytes(&raw).unwrap();
    enc.set_source_size_hint(payload.len() as u64);
    let _ = enc.compress_independent_frame(&payload);
    assert_eq!(enc.state.strategy_tag, StrategyTag::Btlazy2);
}

/// A serialized dictionary (magic, id, entropy tables, repeat offsets,
/// content) around `content`. The entropy section is the fixed blob the
/// dictionary parser tests use.
fn serialized_dictionary(id: u32, content: &[u8]) -> Vec<u8> {
    let mut raw = crate::decoding::dictionary::MAGIC_NUM.to_vec();
    raw.extend_from_slice(&id.to_le_bytes());
    raw.extend_from_slice(&[
        54, 16, 192, 155, 4, 0, 207, 59, 239, 121, 158, 116, 220, 93, 114, 229, 110, 41, 249, 95,
        165, 255, 83, 202, 254, 68, 74, 159, 63, 161, 100, 151, 137, 21, 184, 183, 189, 100, 235,
        209, 251, 174, 91, 75, 91, 185, 19, 39, 75, 146, 98, 177, 249, 14, 4, 35, 0, 0, 0, 40, 40,
        20, 10, 12, 204, 37, 196, 1, 173, 122, 0, 4, 0, 128, 1, 2, 2, 25, 32, 27, 27, 22, 24, 26,
        18, 12, 12, 15, 16, 11, 69, 37, 225, 48, 20, 12, 6, 2, 161, 80, 40, 20, 44, 137, 145, 204,
        46, 0, 0, 0, 0, 0, 116, 253, 16, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
    ]);
    for rep in [1u32, 4, 8] {
        raw.extend_from_slice(&rep.to_le_bytes());
    }
    raw.extend_from_slice(content);
    raw
}

/// Regression: the borrowed (in-place) one-shot path must keep the
/// coordinate floor above every position a previous frame indexed. Zeroing
/// the floor per borrowed frame left the chain / tree tables holding the
/// previous frame's positions as if they were this frame's: a stale entry
/// at the current position was accepted as a match of offset 0 (a corrupt
/// frame), a later one read past the input. Compressing the same input
/// twice on one compressor must give the same decodable frame.
#[test]
fn reused_compressor_borrowed_chain_frames_are_byte_identical() {
    let payload: Vec<u8> = (0..10 * 1024usize)
        .flat_map(|i| format!("row={i} val={}\n", (i * 7919) % 1000).into_bytes())
        .take(10 * 1024)
        .collect();
    // 10 KiB one-shot: window 2^14, so the lazy backend searches the chain.
    let mut enc: FrameCompressor = FrameCompressor::new(super::CompressionLevel::Level(6));
    let frame1 = enc.compress_independent_frame(&payload);
    let frame2 = enc.compress_independent_frame(&payload);
    let frame3 = enc.compress_independent_frame(&payload);
    assert_eq!(
        frame1, frame2,
        "second borrowed frame must not see stale positions"
    );
    assert_eq!(
        frame1, frame3,
        "third borrowed frame must not see stale positions"
    );
    for frame in [&frame1, &frame2, &frame3] {
        let mut decoder = FrameDecoder::new();
        let mut decoded = Vec::with_capacity(payload.len());
        decoder.decode_all_to_vec(frame, &mut decoded).unwrap();
        assert_eq!(decoded, payload);
    }
}

/// A matcher that DOES implement in-place ingest, unlike the built-in driver's
/// Simple backend. `Uncompressed` frames must still round-trip: the level, not
/// the backend, decides whether the staged path is required.
struct InPlaceMatcher {
    buffer: Vec<u8>,
    committed: usize,
    window_size: u64,
}

impl InPlaceMatcher {
    fn new(window_size: u64) -> Self {
        Self {
            buffer: Vec::new(),
            committed: 0,
            window_size,
        }
    }
}

impl Matcher for InPlaceMatcher {
    fn get_next_space(&mut self) -> Vec<u8> {
        vec![0; self.window_size as usize]
    }

    fn get_last_space(&mut self) -> &[u8] {
        &self.buffer[..self.committed]
    }

    fn commit_space(&mut self, space: Vec<u8>) {
        self.buffer = space;
        self.committed = self.buffer.len();
    }

    fn fill_in_place(
        &mut self,
        capacity: usize,
        fill: &mut dyn FnMut(&mut Vec<u8>) -> (usize, bool),
    ) -> Option<(usize, bool)> {
        self.buffer.reserve(capacity);
        Some(fill(&mut self.buffer))
    }

    fn uncommitted_input(&self) -> &[u8] {
        &self.buffer[self.committed..]
    }

    fn commit_filled(&mut self, len: usize) {
        self.committed += len;
    }

    fn skip_matching(&mut self) {}

    fn start_matching(&mut self, mut handle_sequence: impl for<'a> FnMut(Sequence<'a>)) {
        handle_sequence(Sequence::Literals {
            literals: &self.buffer[..self.committed],
        });
    }

    fn reset(&mut self, _level: super::CompressionLevel) {
        self.buffer.clear();
        self.committed = 0;
    }

    fn window_size(&self) -> u64 {
        self.window_size
    }
}

#[test]
fn uncompressed_level_keeps_the_payload_with_an_in_place_matcher() {
    let data = generate_data(0x5eed, 4096);
    let mut out = Vec::new();
    let mut compressor: FrameCompressor<&[u8], &mut Vec<u8>, InPlaceMatcher> =
        FrameCompressor::new_with_matcher(
            InPlaceMatcher::new(1 << 20),
            super::CompressionLevel::Uncompressed,
        );
    compressor.set_source(&data[..]);
    compressor.set_drain(&mut out);
    compressor.compress();

    let mut decoder = FrameDecoder::new();
    let mut decoded = Vec::with_capacity(data.len());
    decoder.decode_all_to_vec(&out, &mut decoded).unwrap();
    assert_eq!(
        decoded, data,
        "Uncompressed frames must carry their payload even when the matcher supports in-place ingest"
    );
}

#[test]
fn dictionary_frame_outgrowing_its_window_stays_decodable() {
    // A dictionary inflates `max_window_size` so the primed bytes stay
    // reachable. Once the frame outgrows the advertised window and those bytes
    // are evicted, that inflation has to be retired, or the matcher keeps
    // admitting matches older than the window the frame header declares and
    // emits an offset no decoder can resolve.
    let dict_raw = include_bytes!("../../../dict_tests/dictionary");
    let dict_for_encoder = crate::decoding::Dictionary::decode_dict(dict_raw).unwrap();
    let dict_for_decoder = crate::decoding::Dictionary::decode_dict(dict_raw).unwrap();

    // Repeats on a 4 KiB period, four times the 1 KiB window below, so every
    // match the matcher can find beyond the window is one the decoder cannot
    // resolve. Far past the window overall, so eviction runs many times over.
    let period = generate_data(0xd1c7, 4 * 1024);
    let mut data = Vec::with_capacity(256 * 1024);
    while data.len() < 256 * 1024 {
        data.extend_from_slice(&period);
    }

    let mut out = Vec::new();
    // Level 3 is Dfast, which reads blocks in place and has no borrowed
    // dictionary scan, so this exercises the in-place commit path.
    let params = crate::encoding::CompressionParameters::builder(super::CompressionLevel::Level(3))
        .window_log(10)
        .build()
        .expect("parameters within bounds");
    let mut compressor = FrameCompressor::new(super::CompressionLevel::Level(3));
    compressor.set_parameters(&params);
    compressor
        .set_dictionary(dict_for_encoder)
        .expect("valid dictionary should attach");
    compressor.set_source(data.as_slice());
    compressor.set_drain(&mut out);
    compressor.compress();

    let mut decoder = FrameDecoder::new();
    decoder.add_dict(dict_for_decoder).unwrap();
    let mut decoded = Vec::with_capacity(data.len());
    decoder
        .decode_all_to_vec(&out, &mut decoded)
        .expect("frame must stay within the window it advertises");
    assert_eq!(decoded, data);
}

/// A prepared dictionary is attached by value, so every frame it primes clones
/// it. Owning its parsed tables outright makes that a copy of the whole
/// dictionary per frame — paid on the path where one dictionary serves many
/// small frames, which is the reason to prepare one at all, and counted by
/// nobody weighing how much memory a run holds. Shared, the clone is a handle.
#[test]
fn a_prepared_dictionary_is_shared_by_its_clones_not_copied() {
    use crate::encoding::EncoderDictionary;

    let content = vec![7u8; 64 * 1024];
    let prepared = EncoderDictionary::from_serialized_or_raw_content(&content)
        .expect("raw content dictionary");
    let alongside = prepared.clone();

    assert_eq!(
        prepared.inner.dict_content.as_ptr(),
        alongside.inner.dict_content.as_ptr(),
        "a clone must point at the same content, not a copy of it"
    );
    assert_eq!(
        prepared.inner.dict_content.len(),
        alongside.inner.dict_content.len(),
        "and see the whole of it"
    );
}

/// The buffer blocks are read into is sized from the caller's size hint, once,
/// rather than grown into a block at a time. A fresh compressor starts with an
/// empty buffer, so a frame that grows into it climbs the doubling ladder and
/// hands the pages back at the end of the frame — measured at level 3 over a
/// 1 MB frame as three growth steps and about 2.4 MB of pages faulted back in
/// per frame, against none for a compressor that sizes the buffer up front.
///
/// The hint reaching this path is advisory (a reader may deliver fewer bytes
/// than promised), which is why it was not sized on before. Reserving on it is
/// bounded twice: the same hint has already sized the window and the tables,
/// and the reservation is clamped to the eviction ceiling the buffer reaches
/// anyway.
/// A window at the format's floor makes the row finder evict and re-index
/// continuously over a frame far larger than the window it may reference, so
/// the frame is written almost entirely from a sliding index. Whatever it
/// emits has to reproduce its input.
#[test]
fn a_frame_far_larger_than_its_window_still_round_trips() {
    use crate::encoding::CompressionParameters;

    // A window far smaller than the input, so the window slides and the
    // absolute cursor is rebased while the frame is still being written.
    let params = CompressionParameters::builder(super::CompressionLevel::Level(5))
        .window_log(10)
        .build()
        .expect("window log 10 is the format's floor");

    let mut data = Vec::with_capacity(512 * 1024);
    let mut state: u64 = 0x2545_F491_4F6C_DD1D;
    while data.len() < 512 * 1024 {
        state = state.wrapping_mul(6364136223846793005).wrapping_add(1);
        // Repeating structure, so the row finder has matches to index and the
        // corrupted cursors would change which ones it finds.
        data.extend_from_slice(&(state >> 32).to_le_bytes()[..4]);
        data.extend_from_slice(b"the same tail every time");
    }

    let mut output: Vec<u8> = Vec::new();
    let mut compressor = FrameCompressor::new(super::CompressionLevel::Level(5));
    compressor.set_parameters(&params);
    compressor.set_source_size_hint(data.len() as u64);
    compressor.set_source(data.as_slice());
    compressor.set_drain(&mut output);
    compressor.compress();

    // Sized up front: the decoder writes into the vector's capacity.
    let mut decoded = Vec::with_capacity(data.len());
    crate::decoding::FrameDecoder::new()
        .decode_all_to_vec(&output, &mut decoded)
        .expect("a frame written over a rebased index must still decode");
    assert_eq!(
        decoded, data,
        "the frame has to reproduce its input after the index was rebased"
    );
}

/// Raw frames are emitted straight from the staged buffer and never consult the
/// match finder, so sizing its history for them holds memory the frame has no
/// use for — a window's worth per frame, and a fresh compressor takes and
/// returns it every time.
#[test]
fn a_raw_frame_reserves_no_matcher_history() {
    let data = vec![0u8; 10];
    let mut output: Vec<u8> = Vec::new();
    let mut compressor = FrameCompressor::new(super::CompressionLevel::Uncompressed);
    compressor.set_source_size_hint(data.len() as u64);
    compressor.set_source(data.as_slice());
    compressor.set_drain(&mut output);
    compressor.compress();

    assert_eq!(
        compressor.state.matcher.ingest_capacity(),
        0,
        "a raw frame reads no history, so none should have been taken for it"
    );
}

/// A dictionary is primed into the ingest buffer before the frame is sized, so
/// a reservation counted from the frame alone leaves the dictionary's own bytes
/// to be grown into afterwards — the doubling chain again, on exactly the path
/// where one dictionary serves many small frames.
#[test]
fn a_dictionary_frame_reserves_room_for_the_dictionary_too() {
    use crate::encoding::EncoderDictionary;

    let dictionary = vec![7u8; 96 * 1024];
    let prepared = EncoderDictionary::from_serialized_or_raw_content(&dictionary)
        .expect("a raw-content dictionary");
    let data = vec![0u8; 700_000];

    for level in [1, 3, 5] {
        let mut output: Vec<u8> = Vec::new();
        let mut compressor = FrameCompressor::new(super::CompressionLevel::Level(level));
        compressor
            .set_encoder_dictionary(prepared.clone())
            .expect("the prepared dictionary attaches");
        compressor.set_source_size_hint(data.len() as u64);
        compressor.set_source(data.as_slice());
        compressor.set_drain(&mut output);
        compressor.compress();

        let capacity = compressor.state.matcher.ingest_capacity();
        let needed = data.len() + dictionary.len();
        assert!(
            capacity >= needed,
            "level {level}: the dictionary and the frame both live in this \
             buffer, so both have to fit: {capacity} < {needed}"
        );
        // And fit by reservation rather than by overshooting into them: the
        // slack is the one block the final top-up asks for, where a buffer that
        // grew lands on a doubling step well past it.
        //
        // Level 1 is excluded from the tight bound: the Fast backend's
        // dictionary is not in the buffer when the frame is sized — priming
        // widens its eviction band by the dictionary's length and the bytes
        // arrive afterwards — so its buffer still ends past the reservation.
        // Left as it is rather than asserted loosely in the other direction,
        // since the reason it lands where it does is not established here.
        if level != 1 {
            assert!(
                capacity <= needed + 256 * 1024,
                "level {level}: {capacity} is past what the frame and \
                 dictionary need ({needed}), which is what growth by doubling \
                 leaves behind"
            );
        }
    }
}

/// A size hint given to the streaming entry point is advisory: the reader may
/// deliver far less than promised. Sizing the ingest buffer from it must not
/// let a wrong number turn into an allocation the data never justifies —
/// the window may be overridden up to a gibibyte, and twice that is the
/// buffer's own ceiling, so an unchecked reservation on a near-empty stream
/// would ask for memory no such stream needs.
#[test]
fn an_overstated_hint_does_not_reserve_what_the_stream_never_delivers() {
    use crate::encoding::CompressionParameters;

    let params = CompressionParameters::builder(super::CompressionLevel::Level(1))
        .window_log(30)
        .build()
        .expect("window log 30 is within the public bounds");

    let mut output: Vec<u8> = Vec::new();
    let mut compressor = FrameCompressor::new(super::CompressionLevel::Level(1));
    compressor.set_parameters(&params);
    // Promised gibibytes, delivers ten bytes.
    compressor.set_source_size_hint(4 * 1024 * 1024 * 1024);
    compressor.set_source(&b"0123456789"[..]);
    compressor.set_drain(&mut output);
    compressor.compress();

    let capacity = compressor.state.matcher.ingest_capacity();
    assert!(
        capacity <= 64 * 1024 * 1024,
        "a hint the stream did not honour must not reserve unbounded memory: \
         {capacity} bytes held for ten"
    );
}

#[test]
fn the_ingest_buffer_is_sized_from_the_hint_not_grown_into() {
    // One level per backend that keeps an ingest buffer: 1 is the Fast
    // matcher, 3 the double-fast, 5 the row finder, 13 its tree, 16 the
    // optimal parser. The size is one no doubling step lands on, so a grown
    // buffer cannot pass by coincidence.
    for level in [1, 3, 5, 13, 16] {
        let data = vec![0u8; 700_000];
        let mut output: Vec<u8> = Vec::new();
        let mut compressor = FrameCompressor::new(super::CompressionLevel::Level(level));
        compressor.set_source_size_hint(data.len() as u64);
        compressor.set_source(data.as_slice());
        compressor.set_drain(&mut output);
        compressor.compress();

        let capacity = compressor.state.matcher.ingest_capacity();
        assert!(
            capacity >= data.len(),
            "level {level}: the whole frame has to fit without growing: \
             {capacity} < {}",
            data.len()
        );
        assert!(
            !capacity.is_power_of_two(),
            "level {level}: a capacity that is an exact power of two is what \
             growth by doubling leaves behind; a reservation lands on the size \
             asked for: {capacity}"
        );
    }
}

/// A prepared dictionary's reported size covers the whole shared allocation:
/// the parts it holds inline, the parsed dictionary's own heap (a fully
/// decoded one carries decode tables) and the encoder entropy tables. The
/// memory queries built on it would otherwise understate what a dictionary
/// pins.
#[test]
fn a_prepared_dictionary_reports_everything_it_holds() {
    let dict_raw = include_bytes!("../../../dict_tests/dictionary");
    let parsed = crate::decoding::Dictionary::decode_dict(dict_raw).unwrap();
    let parsed_heap = parsed.heap_bytes();
    let prepared = super::EncoderDictionary::from_dictionary(parsed);
    let floor = core::mem::size_of::<super::EncoderDictionaryParts>()
        + parsed_heap
        + prepared.inner.entropy.heap_size();
    assert!(
        prepared.heap_size() >= floor,
        "{} < {floor}",
        prepared.heap_size()
    );
}