vcomp_prototype.py

#!/usr/bin/env python3
"""
VCOMP prototype: multilevel lossless compressor with dynamic level-4 labels.

This is an engineering prototype (not "universal eternal compression").
It guarantees lossless roundtrip and uses data-dependent payload strategies.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import mmap
import random
import sqlite3
import struct
import sys
import tempfile
import uuid
import zlib
from concurrent.futures import ProcessPoolExecutor
from collections import deque
from array import array
from datetime import datetime, timezone
from pathlib import Path


MAGIC_V1 = b"VCOMP001"  # old format, no metadata block
MAGIC_V2 = b"VCOMP002"  # current format, metadata block
HEADER_V1_FMT = "<8sBQQ16s"     # magic, mode, original_size, payload_size, checksum16
HEADER_V2_FMT = "<8sBQQI16s"    # + metadata_size (u32)
HEADER_V1_SIZE = struct.calcsize(HEADER_V1_FMT)
HEADER_V2_SIZE = struct.calcsize(HEADER_V2_FMT)
CHECKSUM_OFFSET_V1 = struct.calcsize("<8sBQQ")
CHECKSUM_OFFSET_V2 = struct.calcsize("<8sBQQI")

# Payload modes
MODE_RAW_L4 = 0
MODE_ZLIB_RAW_L4 = 1
MODE_DICT_BITPACK = 2
MODE_ZLIB_DICT_BITPACK = 3
MODE_GLOBAL_DICT = 4
MODE_ZGLOBAL_DICT = 5
MODE_GLOBAL_DICT_MANIFEST = 6
MODE_ZGLOBAL_DICT_MANIFEST = 7
MODE_RANS_RAW_L4 = 8
MODE_RANS_DICT_BITPACK = 9
MODE_RANS_GLOBAL_DICT = 10
MODE_RANS_GLOBAL_DICT_MANIFEST = 11
MODE_RANS_BPE = 12
MODE_RANS_BPE_GLOBAL_DICT = 13
MODE_RANS_BPE_GLOBAL_DICT_MANIFEST = 14

MODE_NAME_TO_ID = {
    "raw": MODE_RAW_L4,
    "zraw": MODE_ZLIB_RAW_L4,
    "dict": MODE_DICT_BITPACK,
    "zdict": MODE_ZLIB_DICT_BITPACK,
    "gdict": MODE_GLOBAL_DICT,
    "zgdict": MODE_ZGLOBAL_DICT,
    "gdictm": MODE_GLOBAL_DICT_MANIFEST,
    "zgdictm": MODE_ZGLOBAL_DICT_MANIFEST,
    "ansraw": MODE_RANS_RAW_L4,
    "ansdict": MODE_RANS_DICT_BITPACK,
    "ansgdict": MODE_RANS_GLOBAL_DICT,
    "ansgdictm": MODE_RANS_GLOBAL_DICT_MANIFEST,
    "ansbpe": MODE_RANS_BPE,
    "ansbpegdict": MODE_RANS_BPE_GLOBAL_DICT,
    "ansbpegdictm": MODE_RANS_BPE_GLOBAL_DICT_MANIFEST,
}
MODE_ID_TO_NAME = {v: k for k, v in MODE_NAME_TO_ID.items()}
GLOBAL_DICT_SCHEMA_VERSION = 1
RANS_SCALE_BITS = 12
RANS_TOTAL_FREQ = 1 << RANS_SCALE_BITS
RANS_L = 1 << 23
RANS_FRAME_MAGIC = b"RANSBLK1"
RANS_FRAME_HEADER_FMT = "<8sI"
RANS_FRAME_BLOCK_FMT = "<II"
RANS_FRAME_HEADER_SIZE = struct.calcsize(RANS_FRAME_HEADER_FMT)
RANS_FRAME_BLOCK_SIZE = struct.calcsize(RANS_FRAME_BLOCK_FMT)
RANS_DEFAULT_BLOCK_BYTES = 1 << 20
BPE_PAYLOAD_HEADER_FMT = "<IIIBI"
BPE_PAYLOAD_HEADER_SIZE = struct.calcsize(BPE_PAYLOAD_HEADER_FMT)
BPE_MAX_MERGES = 128
BPE_MIN_PAIR_COUNT = 4
BPE_GDICTM_PAYLOAD_HEADER_FMT = "<II"
BPE_GDICTM_PAYLOAD_HEADER_SIZE = struct.calcsize(BPE_GDICTM_PAYLOAD_HEADER_FMT)
GLOBAL_BPE_SEED_LIMIT = 64
GLOBAL_BPE_LEARN_LIMIT = 128
GLOBAL_BPE_PAIR_LIMIT = 8192
GLOBAL_BPE_VOCAB_LIMIT = 1024


def level_counts(original_size: int) -> tuple[int, int, int, int, int]:
    """
    Returns (n0, n1, n2, n3, n4) symbol counts for levels 0..4.
    """
    n0 = original_size * 4  # 2-bit symbols
    n1 = n0 // 2
    n2 = n1 // 2
    n3 = (n2 + 1) // 2      # 16-bit symbols
    n4 = (n3 + 1) // 2      # 32-bit symbols
    return n0, n1, n2, n3, n4


def mode_name(mode_id: int) -> str:
    return MODE_ID_TO_NAME.get(mode_id, f"unknown({mode_id})")


def serialize_metadata(metadata: dict | None) -> bytes:
    if metadata is None:
        metadata = {}
    return json.dumps(metadata, separators=(",", ":"), ensure_ascii=False).encode("utf-8")


def deserialize_metadata(blob: bytes) -> dict:
    if not blob:
        return {}
    try:
        obj = json.loads(blob.decode("utf-8"))
        return obj if isinstance(obj, dict) else {}
    except Exception:
        return {}


def detect_image_dimensions(path: Path) -> tuple[str, int, int] | None:
    try:
        with path.open("rb") as f:
            sig = f.read(32)
            if sig.startswith(b"\x89PNG\r\n\x1a\n") and len(sig) >= 24:
                w = struct.unpack(">I", sig[16:20])[0]
                h = struct.unpack(">I", sig[20:24])[0]
                return ("png", w, h)

            if sig.startswith((b"GIF87a", b"GIF89a")) and len(sig) >= 10:
                w = struct.unpack("<H", sig[6:8])[0]
                h = struct.unpack("<H", sig[8:10])[0]
                return ("gif", w, h)

            if sig.startswith(b"BM"):
                if len(sig) < 26:
                    f.seek(0)
                    sig = f.read(64)
                if len(sig) >= 26:
                    w = struct.unpack("<I", sig[18:22])[0]
                    h_signed = struct.unpack("<i", sig[22:26])[0]
                    return ("bmp", w, abs(h_signed))

            f.seek(0)
            if f.read(2) == b"\xFF\xD8":
                sof_markers = {
                    0xC0, 0xC1, 0xC2, 0xC3,
                    0xC5, 0xC6, 0xC7,
                    0xC9, 0xCA, 0xCB,
                    0xCD, 0xCE, 0xCF,
                }
                while True:
                    marker_prefix = f.read(1)
                    if not marker_prefix:
                        break
                    if marker_prefix != b"\xFF":
                        continue

                    marker = f.read(1)
                    while marker == b"\xFF":
                        marker = f.read(1)
                    if not marker:
                        break
                    m = marker[0]

                    if m in (0xD8, 0xD9) or (0xD0 <= m <= 0xD7) or m == 0x01:
                        continue

                    seg_len_bytes = f.read(2)
                    if len(seg_len_bytes) != 2:
                        break
                    seg_len = struct.unpack(">H", seg_len_bytes)[0]
                    if seg_len < 2:
                        break

                    if m in sof_markers:
                        seg = f.read(seg_len - 2)
                        if len(seg) >= 5:
                            h = struct.unpack(">H", seg[1:3])[0]
                            w = struct.unpack(">H", seg[3:5])[0]
                            return ("jpeg", w, h)
                        break
                    f.seek(seg_len - 2, 1)
    except Exception:
        return None
    return None


def build_source_metadata(input_path: Path, original_size: int) -> dict:
    meta = {
        "source_name": input_path.name,
        "source_ext": input_path.suffix.lower(),
        "source_size": int(original_size),
    }
    img = detect_image_dimensions(input_path)
    if img is not None:
        fmt, w, h = img
        meta["image"] = {"format": fmt, "width": int(w), "height": int(h)}
    return meta


def pair_symbols(seq: list[int], base: int) -> list[int]:
    out: list[int] = []
    n = len(seq)
    i = 0
    while i < n:
        a = seq[i]
        b = seq[i + 1] if i + 1 < n else 0
        out.append(a * base + b)
        i += 2
    return out


def unpair_symbols(seq: list[int], base: int, target_len: int) -> list[int]:
    out: list[int] = []
    out_extend = out.extend
    for value in seq:
        a = value // base
        b = value % base
        out_extend((a, b))
    if len(out) > target_len:
        out = out[:target_len]
    return out


def build_level4_labels(data: bytes) -> list[int]:
    """
    L2 (bytes) -> L3 (u16-like) -> L4 (u32-like).
    L0/L1 are implied by this reversible chain.
    """
    level2 = list(data)
    level3 = pair_symbols(level2, 256)
    level4 = pair_symbols(level3, 65536)
    return level4


def restore_bytes_from_level4(level4: list[int], original_size: int) -> bytes:
    _, _, n2, n3, _ = level_counts(original_size)
    level3 = unpair_symbols(level4, 65536, n3)
    level2 = unpair_symbols(level3, 256, n2)
    return bytes(level2)


def pack_u32_list(values: list[int]) -> bytes:
    arr = array("I", values)
    if sys.byteorder != "little":
        arr.byteswap()
    return arr.tobytes()


def unpack_u32_list(blob: bytes) -> list[int]:
    if len(blob) % 4 != 0:
        raise ValueError("Invalid u32 blob length")
    arr = array("I")
    arr.frombytes(blob)
    if sys.byteorder != "little":
        arr.byteswap()
    return arr.tolist()


def pack_u32_array(values: array) -> bytes:
    if values.typecode != "I":
        raise ValueError("Expected array('I')")
    if sys.byteorder == "little":
        return values.tobytes()
    tmp = array("I", values)
    tmp.byteswap()
    return tmp.tobytes()


def bits_required(num_values: int) -> int:
    if num_values <= 1:
        return 1
    return (num_values - 1).bit_length()


def bitpack(values: list[int], bit_width: int) -> bytes:
    if not (1 <= bit_width <= 32):
        raise ValueError(f"Invalid bit width: {bit_width}")

    out = bytearray()
    buf = 0
    bits = 0
    mask = (1 << bit_width) - 1
    for v in values:
        buf |= (v & mask) << bits
        bits += bit_width
        while bits >= 8:
            out.append(buf & 0xFF)
            buf >>= 8
            bits -= 8
    if bits:
        out.append(buf & 0xFF)
    return bytes(out)


def bitunpack(blob: bytes, count: int, bit_width: int) -> list[int]:
    if not (1 <= bit_width <= 32):
        raise ValueError(f"Invalid bit width: {bit_width}")

    out: list[int] = []
    buf = 0
    bits = 0
    mask = (1 << bit_width) - 1
    for b in blob:
        buf |= b << bits
        bits += 8
        while bits >= bit_width and len(out) < count:
            out.append(buf & mask)
            buf >>= bit_width
            bits -= bit_width
    if len(out) != count:
        raise ValueError("Bit-unpack failed: not enough data")
    return out


def bitpack_u32_file_to_writer(
    idx_file,
    index_count: int,
    bit_width: int,
    write_bytes,
    chunk_words: int = 1 << 16,
) -> int:
    """
    Stream-pack u32 indices from file into fixed-width bitstream.
    Returns bytes written.
    """
    if not (1 <= bit_width <= 32):
        raise ValueError(f"Invalid bit width: {bit_width}")

    idx_file.seek(0)
    total_written = 0
    produced = 0
    bitbuf = 0
    bits = 0
    mask = (1 << bit_width) - 1
    out = bytearray()

    while produced < index_count:
        need = min(chunk_words, index_count - produced)
        raw = idx_file.read(need * 4)
        if len(raw) != need * 4:
            raise ValueError("Corrupted temporary index stream")

        arr = array("I")
        arr.frombytes(raw)
        if sys.byteorder != "little":
            arr.byteswap()

        for v in arr:
            bitbuf |= (v & mask) << bits
            bits += bit_width
            while bits >= 8:
                out.append(bitbuf & 0xFF)
                bitbuf >>= 8
                bits -= 8
                if len(out) >= (1 << 20):
                    write_bytes(out)
                    total_written += len(out)
                    out.clear()
        produced += len(arr)

    if bits:
        out.append(bitbuf & 0xFF)
    if out:
        write_bytes(out)
        total_written += len(out)
    return total_written


def _read_exact(stream, size: int) -> bytes:
    data = stream.read(size)
    if len(data) != size:
        raise ValueError("Unexpected end of stream")
    return data


def _pack_big_endian_u32_chunk(chunk: bytes) -> bytes:
    """
    Convert 4-byte aligned input chunk into little-endian u32 payload bytes,
    where each u32 is interpreted from big-endian source order.
    """
    if len(chunk) % 4 != 0:
        raise ValueError("Chunk must be aligned to 4 bytes")
    arr = array("I", (word[0] for word in struct.iter_unpack(">I", chunk)))
    if sys.byteorder != "little":
        arr.byteswap()
    return arr.tobytes()


def _iter_l4_raw_chunks(input_stream, chunk_size: int, workers: int, hasher):
    """
    Yield tuples: (raw_l4_bytes_le, l4_word_count) in strict input order.
    raw_l4_bytes_le is a contiguous little-endian u32 stream.
    """
    workers = max(1, int(workers))

    if workers == 1:
        encoder = L4EncoderState()
        while True:
            chunk = input_stream.read(chunk_size)
            if not chunk:
                break
            hasher.update(chunk)
            words = encoder.feed(chunk)
            if words:
                yield pack_u32_array(words), len(words)

        tail_words = encoder.finalize()
        if tail_words:
            yield pack_u32_array(tail_words), len(tail_words)
        return

    if chunk_size < 4:
        raise ValueError("For workers>1 chunk_size must be >= 4")

    pending = deque()
    carry = b""
    max_pending = max(2, workers * 2)

    with ProcessPoolExecutor(max_workers=workers) as pool:
        while True:
            chunk = input_stream.read(chunk_size)
            if not chunk:
                break
            hasher.update(chunk)

            buf = carry + chunk
            aligned = (len(buf) // 4) * 4
            if aligned:
                piece = bytes(buf[:aligned])
                future = pool.submit(_pack_big_endian_u32_chunk, piece)
                pending.append((future, aligned // 4))
                carry = buf[aligned:]
            else:
                carry = buf

            while len(pending) >= max_pending:
                fut, words_cnt = pending.popleft()
                yield fut.result(), words_cnt

        if carry:
            pad = carry + b"\x00" * (4 - len(carry))
            future = pool.submit(_pack_big_endian_u32_chunk, pad)
            pending.append((future, 1))

        while pending:
            fut, words_cnt = pending.popleft()
            yield fut.result(), words_cnt


def encode_dict_bitpack(level4: list[int]) -> bytes:
    """
    Encode level4 labels using:
    - dynamic dictionary (only seen labels)
    - minimal-bit packed index stream
    """
    dictionary: list[int] = []
    lookup: dict[int, int] = {}
    indices: list[int] = []

    for label in level4:
        idx = lookup.get(label)
        if idx is None:
            idx = len(dictionary)
            lookup[label] = idx
            dictionary.append(label)
        indices.append(idx)

    dict_count = len(dictionary)
    bit_width = bits_required(dict_count)
    packed_idx = bitpack(indices, bit_width)
    dict_blob = pack_u32_list(dictionary)

    # <I B I I> dict_count, bit_width, index_count, packed_size
    head = struct.pack("<IBII", dict_count, bit_width, len(indices), len(packed_idx))
    return head + dict_blob + packed_idx


def decode_dict_bitpack(blob: bytes) -> list[int]:
    head_size = struct.calcsize("<IBII")
    if len(blob) < head_size:
        raise ValueError("Invalid dict-bitpack payload (too short)")

    dict_count, bit_width, index_count, packed_size = struct.unpack("<IBII", blob[:head_size])
    if not (1 <= bit_width <= 32):
        raise ValueError("Invalid dict-bitpack payload (bad bit width)")
    if dict_count == 0 and index_count > 0:
        raise ValueError("Invalid dict-bitpack payload (empty dict with non-empty index stream)")

    pos = head_size
    dict_bytes = dict_count * 4
    if len(blob) != pos + dict_bytes + packed_size:
        raise ValueError("Invalid dict-bitpack payload (size mismatch)")

    dictionary = unpack_u32_list(blob[pos : pos + dict_bytes])
    pos += dict_bytes
    packed_idx = blob[pos : pos + packed_size]
    indices = bitunpack(packed_idx, index_count, bit_width)

    try:
        return [dictionary[i] for i in indices]
    except IndexError as exc:
        raise ValueError("Corrupted index stream: dictionary index out of range") from exc


def decode_global_manifest_bitpack(blob: bytes) -> list[int]:
    head_size = struct.calcsize("<III")
    if len(blob) < head_size:
        raise ValueError("Invalid global-manifest payload (too short)")

    manifest_count, index_count, packed_size = struct.unpack("<III", blob[:head_size])
    pos = head_size
    manifest_bytes = manifest_count * 8
    if len(blob) != pos + manifest_bytes + packed_size:
        raise ValueError("Invalid global-manifest payload (size mismatch)")

    manifest: dict[int, int] = {}
    max_gid = 0
    end_manifest = pos + manifest_bytes
    while pos < end_manifest:
        gid, label = struct.unpack("<II", blob[pos : pos + 8])
        if gid in manifest:
            raise ValueError("Invalid global-manifest payload (duplicate gid)")
        manifest[int(gid)] = int(label)
        if gid > max_gid:
            max_gid = int(gid)
        pos += 8

    if not manifest and index_count > 0:
        raise ValueError("Invalid global-manifest payload (empty manifest with non-empty index stream)")

    bit_width = max(1, max_gid.bit_length())
    expected_packed_size = (index_count * bit_width + 7) // 8
    if packed_size != expected_packed_size:
        raise ValueError("Invalid global-manifest payload (packed size mismatch)")

    gids = bitunpack(blob[pos : pos + packed_size], index_count, bit_width)
    try:
        return [manifest[int(gid)] for gid in gids]
    except KeyError as exc:
        raise ValueError("Corrupted global-manifest payload: gid missing from manifest") from exc


def _build_rans_frequencies(data: bytes) -> tuple[dict[int, int], dict[int, int]]:
    counts = [0] * 256
    for b in data:
        counts[b] += 1

    nonzero = [(sym, cnt) for sym, cnt in enumerate(counts) if cnt > 0]
    if not nonzero:
        return {}, {}
    if len(nonzero) == 1:
        sym = nonzero[0][0]
        return {sym: RANS_TOTAL_FREQ}, {sym: 0}

    total = len(data)
    freqs: dict[int, int] = {}
    for sym, cnt in nonzero:
        freq = (cnt * RANS_TOTAL_FREQ) // total
        if freq <= 0:
            freq = 1
        freqs[sym] = int(freq)

    freq_sum = sum(freqs.values())
    if freq_sum < RANS_TOTAL_FREQ:
        best_sym = max(nonzero, key=lambda item: (item[1], -item[0]))[0]
        freqs[best_sym] += RANS_TOTAL_FREQ - freq_sum
    elif freq_sum > RANS_TOTAL_FREQ:
        overflow = freq_sum - RANS_TOTAL_FREQ
        for sym, _ in sorted(nonzero, key=lambda item: (-freqs[item[0]], item[0])):
            if overflow <= 0:
                break
            removable = min(overflow, freqs[sym] - 1)
            if removable > 0:
                freqs[sym] -= removable
                overflow -= removable
        if overflow != 0:
            raise ValueError("Failed to normalize rANS frequencies")

    starts: dict[int, int] = {}
    cum = 0
    for sym in sorted(freqs):
        starts[sym] = cum
        cum += freqs[sym]
    if cum != RANS_TOTAL_FREQ:
        raise ValueError("Invalid rANS normalization total")
    return freqs, starts


def _iter_file_bytes_reversed(path: Path, chunk_size: int = 1 << 20):
    with path.open("rb") as f:
        pos = path.stat().st_size
        while pos > 0:
            take = min(chunk_size, pos)
            pos -= take
            f.seek(pos)
            chunk = f.read(take)
            for b in reversed(chunk):
                yield b


def _build_rans_frequencies_from_file(path: Path, chunk_size: int = 1 << 20) -> tuple[int, dict[int, int], dict[int, int]]:
    counts = [0] * 256
    total = 0
    with path.open("rb") as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            total += len(chunk)
            for b in chunk:
                counts[b] += 1

    nonzero = [(sym, cnt) for sym, cnt in enumerate(counts) if cnt > 0]
    if not nonzero:
        return total, {}, {}
    if len(nonzero) == 1:
        sym = nonzero[0][0]
        return total, {sym: RANS_TOTAL_FREQ}, {sym: 0}

    freqs: dict[int, int] = {}
    for sym, cnt in nonzero:
        freq = (cnt * RANS_TOTAL_FREQ) // total
        if freq <= 0:
            freq = 1
        freqs[sym] = int(freq)

    freq_sum = sum(freqs.values())
    if freq_sum < RANS_TOTAL_FREQ:
        best_sym = max(nonzero, key=lambda item: (item[1], -item[0]))[0]
        freqs[best_sym] += RANS_TOTAL_FREQ - freq_sum
    elif freq_sum > RANS_TOTAL_FREQ:
        overflow = freq_sum - RANS_TOTAL_FREQ
        for sym, _ in sorted(nonzero, key=lambda item: (-freqs[item[0]], item[0])):
            if overflow <= 0:
                break
            removable = min(overflow, freqs[sym] - 1)
            if removable > 0:
                freqs[sym] -= removable
                overflow -= removable
        if overflow != 0:
            raise ValueError("Failed to normalize rANS frequencies")

    starts: dict[int, int] = {}
    cum = 0
    for sym in sorted(freqs):
        starts[sym] = cum
        cum += freqs[sym]
    if cum != RANS_TOTAL_FREQ:
        raise ValueError("Invalid rANS normalization total")
    return total, freqs, starts


def rans_encode_bytes(data: bytes) -> bytes:
    head = bytearray()
    head.extend(struct.pack("<QH", len(data), 0))
    if not data:
        return bytes(head)

    freqs, starts = _build_rans_frequencies(data)
    head = bytearray()
    head.extend(struct.pack("<QH", len(data), len(freqs)))
    for sym in sorted(freqs):
        head.extend(struct.pack("<BH", sym, freqs[sym]))

    state = RANS_L
    out = bytearray()
    x_max_base = (RANS_L >> RANS_SCALE_BITS) << 8
    for sym in reversed(data):
        freq = freqs[sym]
        x_max = x_max_base * freq
        while state >= x_max:
            out.append(state & 0xFF)
            state >>= 8
        start = starts[sym]
        state = ((state // freq) << RANS_SCALE_BITS) + (state % freq) + start

    head.extend(struct.pack("<I", state))
    head.extend(reversed(out))
    return bytes(head)


def _rans_encode_block_worker(raw_block: bytes) -> bytes:
    return rans_encode_bytes(raw_block)


def rans_decode_bytes(blob: bytes) -> bytes:
    head_size = struct.calcsize("<QH")
    if len(blob) < head_size:
        raise ValueError("Invalid rANS payload (too short)")

    raw_size, symbol_count = struct.unpack("<QH", blob[:head_size])
    pos = head_size
    if symbol_count == 0:
        if raw_size != 0 or len(blob) != head_size:
            raise ValueError("Invalid rANS payload (empty table mismatch)")
        return b""

    table_size = symbol_count * struct.calcsize("<BH")
    if len(blob) < pos + table_size + 4:
        raise ValueError("Invalid rANS payload (truncated header)")

    freqs: dict[int, int] = {}
    starts: dict[int, int] = {}
    decode_table = bytearray(RANS_TOTAL_FREQ)
    cum = 0
    for _ in range(symbol_count):
        sym, freq = struct.unpack("<BH", blob[pos : pos + 3])
        pos += 3
        if freq <= 0:
            raise ValueError("Invalid rANS payload (zero frequency)")
        if sym in freqs:
            raise ValueError("Invalid rANS payload (duplicate symbol)")
        freqs[int(sym)] = int(freq)
        starts[int(sym)] = cum
        decode_table[cum : cum + freq] = bytes([sym]) * freq
        cum += freq
    if cum != RANS_TOTAL_FREQ:
        raise ValueError("Invalid rANS payload (frequency total mismatch)")

    state = struct.unpack("<I", blob[pos : pos + 4])[0]
    pos += 4
    stream = blob[pos:]
    stream_pos = 0

    out = bytearray()
    for i in range(raw_size):
        slot = state & (RANS_TOTAL_FREQ - 1)
        sym = decode_table[slot]
        out.append(sym)
        state = freqs[sym] * (state >> RANS_SCALE_BITS) + (slot - starts[sym])
        while state < RANS_L and stream_pos < len(stream):
            state = (state << 8) | stream[stream_pos]
            stream_pos += 1
        if state < RANS_L and stream_pos >= len(stream) and i + 1 < raw_size:
            raise ValueError("Invalid rANS payload (unexpected end of renormalization stream)")

    if stream_pos != len(stream):
        raise ValueError("Invalid rANS payload (trailing bytes)")
    return bytes(out)


def _rans_decode_block_worker(encoded_block: bytes) -> bytes:
    return rans_decode_bytes(encoded_block)


class FramedRansEncoder:
    def __init__(self, write_bytes, block_size: int = RANS_DEFAULT_BLOCK_BYTES):
        if block_size <= 0:
            raise ValueError("rANS framed block size must be positive")
        self.write_bytes = write_bytes
        self.block_size = int(block_size)
        self._buf = bytearray()
        self._header_written = False
        self._closed = False
        self._total = 0
        self._block_count = 0

    def _write_header(self) -> None:
        if self._header_written:
            return
        header = struct.pack(RANS_FRAME_HEADER_FMT, RANS_FRAME_MAGIC, self.block_size)
        self.write_bytes(header)
        self._total += len(header)
        self._header_written = True

    def _emit_block(self, raw_block: bytes) -> None:
        encoded = rans_encode_bytes(raw_block)
        block_head = struct.pack(RANS_FRAME_BLOCK_FMT, len(raw_block), len(encoded))
        self.write_bytes(block_head)
        self.write_bytes(encoded)
        self._total += len(block_head) + len(encoded)
        self._block_count += 1

    def feed(self, data: bytes | memoryview) -> None:
        if self._closed:
            raise ValueError("Framed rANS encoder already closed")
        if not data:
            return
        self._write_header()
        self._buf.extend(data)
        while len(self._buf) >= self.block_size:
            raw_block = bytes(self._buf[: self.block_size])
            del self._buf[: self.block_size]
            self._emit_block(raw_block)

    def finish(self) -> int:
        if self._closed:
            raise ValueError("Framed rANS encoder already closed")
        self._write_header()
        if self._buf:
            self._emit_block(bytes(self._buf))
            self._buf.clear()
        self._closed = True
        return self._total


def rans_encode_framed_bytes(
    data: bytes,
    block_size: int = RANS_DEFAULT_BLOCK_BYTES,
    workers: int = 1,
) -> bytes:
    workers = max(1, int(workers))
    if block_size <= 0:
        raise ValueError("rANS framed block size must be positive")

    if workers <= 1 or len(data) <= block_size:
        out = bytearray()
        encoder = FramedRansEncoder(out.extend, block_size=block_size)
        encoder.feed(data)
        encoder.finish()
        return bytes(out)

    blocks = [data[pos : pos + block_size] for pos in range(0, len(data), block_size)]
    out = bytearray(struct.pack(RANS_FRAME_HEADER_FMT, RANS_FRAME_MAGIC, block_size))
    with ProcessPoolExecutor(max_workers=min(workers, len(blocks))) as ex:
        encoded_blocks = list(ex.map(_rans_encode_block_worker, blocks))

    for raw_block, encoded in zip(blocks, encoded_blocks):
        out.extend(struct.pack(RANS_FRAME_BLOCK_FMT, len(raw_block), len(encoded)))
        out.extend(encoded)
    return bytes(out)


def _parse_framed_rans_blocks(blob: bytes) -> tuple[int, list[tuple[int, bytes]]]:
    if len(blob) < RANS_FRAME_HEADER_SIZE:
        raise ValueError("Invalid framed rANS payload (too short)")

    magic, block_size = struct.unpack(RANS_FRAME_HEADER_FMT, blob[:RANS_FRAME_HEADER_SIZE])
    if magic != RANS_FRAME_MAGIC:
        raise ValueError("Invalid framed rANS payload (bad magic)")
    if block_size <= 0:
        raise ValueError("Invalid framed rANS payload (bad block size)")

    pos = RANS_FRAME_HEADER_SIZE
    blocks: list[tuple[int, bytes]] = []
    while pos < len(blob):
        if len(blob) < pos + RANS_FRAME_BLOCK_SIZE:
            raise ValueError("Invalid framed rANS payload (truncated block header)")
        raw_size, encoded_size = struct.unpack(
            RANS_FRAME_BLOCK_FMT,
            blob[pos : pos + RANS_FRAME_BLOCK_SIZE],
        )
        pos += RANS_FRAME_BLOCK_SIZE
        if len(blob) < pos + encoded_size:
            raise ValueError("Invalid framed rANS payload (truncated block)")
        blocks.append((raw_size, blob[pos : pos + encoded_size]))
        pos += encoded_size
    return block_size, blocks


def rans_decode_framed_bytes(blob: bytes, workers: int = 1) -> bytes:
    _, blocks = _parse_framed_rans_blocks(blob)
    if not blocks:
        return b""

    workers = max(1, int(workers))
    encoded_blocks = [encoded for _, encoded in blocks]
    raw_sizes = [raw_size for raw_size, _ in blocks]

    if workers <= 1 or len(blocks) <= 1:
        decoded_blocks = [rans_decode_bytes(encoded) for encoded in encoded_blocks]
    else:
        with ProcessPoolExecutor(max_workers=min(workers, len(blocks))) as ex:
            decoded_blocks = list(ex.map(_rans_decode_block_worker, encoded_blocks))

    out = bytearray()
    for raw_size, raw in zip(raw_sizes, decoded_blocks):
        if len(raw) != raw_size:
            raise ValueError("Invalid framed rANS payload (block size mismatch)")
        out.extend(raw)
    return bytes(out)


def rans_decode_auto_bytes(blob: bytes, workers: int = 1) -> bytes:
    if blob.startswith(RANS_FRAME_MAGIC):
        return rans_decode_framed_bytes(blob, workers=workers)
    return rans_decode_bytes(blob)


def _reverse_copy_file_to_writer(src_path: Path, write_bytes, chunk_size: int = 1 << 20) -> int:
    total = 0
    with src_path.open("rb") as f:
        pos = src_path.stat().st_size
        while pos > 0:
            take = min(chunk_size, pos)
            pos -= take
            f.seek(pos)
            chunk = f.read(take)
            if chunk:
                rev = chunk[::-1]
                write_bytes(rev)
                total += len(rev)
    return total


def rans_encode_file_to_writer(
    input_path: Path,
    write_bytes,
    scratch_dir: Path | None = None,
    io_chunk_size: int = 1 << 20,
) -> int:
    raw_size, freqs, starts = _build_rans_frequencies_from_file(input_path, chunk_size=io_chunk_size)
    header = bytearray()
    header.extend(struct.pack("<QH", raw_size, len(freqs)))
    if not freqs:
        write_bytes(header)
        return len(header)

    for sym in sorted(freqs):
        header.extend(struct.pack("<BH", sym, freqs[sym]))

    with tempfile.TemporaryDirectory(prefix="vcomp_rans_emit_", dir=scratch_dir) as td:
        emit_path = Path(td) / "emit.bin"
        state = RANS_L
        x_max_base = (RANS_L >> RANS_SCALE_BITS) << 8
        emitted_count = 0

        with emit_path.open("wb") as emit_f:
            for sym in _iter_file_bytes_reversed(input_path, chunk_size=io_chunk_size):
                freq = freqs[sym]
                x_max = x_max_base * freq
                while state >= x_max:
                    emit_f.write(bytes([state & 0xFF]))
                    emitted_count += 1
                    state >>= 8
                start = starts[sym]
                state = ((state // freq) << RANS_SCALE_BITS) + (state % freq) + start

        write_bytes(header)
        write_bytes(struct.pack("<I", state))
        total = len(header) + 4
        total += _reverse_copy_file_to_writer(emit_path, write_bytes, chunk_size=io_chunk_size)
        if total != len(header) + 4 + emitted_count:
            raise ValueError("Internal error: rANS file encoding size mismatch")
        return total


def rans_encode_framed_file_to_writer(
    input_path: Path,
    write_bytes,
    block_size: int = RANS_DEFAULT_BLOCK_BYTES,
    io_chunk_size: int = 1 << 20,
) -> int:
    encoder = FramedRansEncoder(write_bytes, block_size=block_size)
    with input_path.open("rb") as in_f:
        while True:
            chunk = in_f.read(io_chunk_size)
            if not chunk:
                break
            encoder.feed(chunk)
    return encoder.finish()


def _emit_temp_payload_file(
    raw_payload_path: Path,
    out_f,
    entropy_mode: str,
    io_chunk_size: int = 1 << 20,
) -> int:
    if entropy_mode == "raw":
        total = 0
        with raw_payload_path.open("rb") as raw_f:
            while True:
                chunk = raw_f.read(io_chunk_size)
                if not chunk:
                    break
                out_f.write(chunk)
                total += len(chunk)
        return total

    if entropy_mode == "zlib":
        total = 0
        compressor = zlib.compressobj(level=9)
        with raw_payload_path.open("rb") as raw_f:
            while True:
                chunk = raw_f.read(io_chunk_size)
                if not chunk:
                    break
                comp = compressor.compress(chunk)
                if comp:
                    out_f.write(comp)
                    total += len(comp)
        tail = compressor.flush()
        if tail:
            out_f.write(tail)
            total += len(tail)
        return total

    if entropy_mode == "rans":
        return rans_encode_framed_file_to_writer(
            raw_payload_path,
            out_f.write,
            block_size=max(1 << 16, io_chunk_size),
            io_chunk_size=io_chunk_size,
        )

    raise ValueError(f"Unknown payload entropy mode: {entropy_mode}")


def _read_container_payload_bytes(input_path: Path, payload_offset: int, payload_size: int) -> bytes:
    with input_path.open("rb") as in_f:
        in_f.seek(payload_offset)
        payload = in_f.read(payload_size)
    if len(payload) != payload_size:
        raise ValueError("Invalid container: payload truncated")
    return payload


def _count_adjacent_pairs(seq: list[int]) -> dict[tuple[int, int], int]:
    if len(seq) < 2:
        return {}
    counts: dict[tuple[int, int], int] = {}
    prev = int(seq[0])
    for cur_raw in seq[1:]:
        cur = int(cur_raw)
        pair = (prev, cur)
        counts[pair] = counts.get(pair, 0) + 1
        prev = cur
    return counts


def _apply_bpe_merge_once(seq: list[int], left: int, right: int, new_id: int) -> tuple[list[int], int]:
    replaced = 0
    out: list[int] = []
    i = 0
    n = len(seq)
    while i < n:
        if i + 1 < n and seq[i] == left and seq[i + 1] == right:
            out.append(new_id)
            replaced += 1
            i += 2
        else:
            out.append(seq[i])
            i += 1
    return out, replaced


def _build_bpe_tokenization(
    level4: list[int],
    max_merges: int = BPE_MAX_MERGES,
    min_pair_count: int = BPE_MIN_PAIR_COUNT,
    seed_pairs: list[tuple[int, int]] | None = None,
) -> tuple[list[int], list[tuple[int, int]], list[int], int]:
    label_to_base: dict[int, int] = {}
    base_labels: list[int] = []
    seq: list[int] = []

    for label in level4:
        base_id = label_to_base.get(label)
        if base_id is None:
            base_id = len(base_labels)
            label_to_base[label] = base_id
            base_labels.append(label)
        seq.append(base_id)

    if len(seq) < 2 or max_merges <= 0:
        return base_labels, [], seq, 0

    merges: list[tuple[int, int]] = []
    next_symbol = len(base_labels)
    seed_applied = 0

    if seed_pairs:
        for left_label, right_label in seed_pairs:
            left = label_to_base.get(int(left_label))
            right = label_to_base.get(int(right_label))
            if left is None or right is None:
                continue
            out, replaced = _apply_bpe_merge_once(seq, left, right, next_symbol)
            if replaced <= 0:
                continue
            merges.append((left, right))
            seq = out
            next_symbol += 1
            seed_applied += 1
            if len(merges) >= max_merges or len(seq) < 2:
                return base_labels, merges, seq, seed_applied

    while len(merges) < max_merges and len(seq) >= 2:
        pair_counts = _count_adjacent_pairs(seq)

        best_pair: tuple[int, int] | None = None
        best_count = min_pair_count - 1
        for pair, count in pair_counts.items():
            if count > best_count or (count == best_count and best_pair is not None and pair < best_pair):
                best_pair = pair
                best_count = count
            elif count > best_count:
                best_pair = pair
                best_count = count

        if best_pair is None or best_count < min_pair_count:
            break

        left, right = best_pair
        out, replaced = _apply_bpe_merge_once(seq, left, right, next_symbol)

        if replaced == 0:
            break

        merges.append(best_pair)
        seq = out
        next_symbol += 1

    return base_labels, merges, seq, seed_applied


def encode_bpeans_sequence(
    symbols: list[int],
    max_merges: int = BPE_MAX_MERGES,
    min_pair_count: int = BPE_MIN_PAIR_COUNT,
    seed_pairs: list[tuple[int, int]] | None = None,
    workers: int = 1,
) -> tuple[bytes, dict]:
    base_labels, merges, token_seq, seed_applied = _build_bpe_tokenization(
        symbols,
        max_merges=max_merges,
        min_pair_count=min_pair_count,
        seed_pairs=seed_pairs,
    )

    vocab_size = len(base_labels) + len(merges)
    bit_width = bits_required(vocab_size if vocab_size > 0 else 1)
    packed_tokens = bitpack(token_seq, bit_width) if token_seq else b""
    packed_size = len(packed_tokens)

    header = struct.pack(
        BPE_PAYLOAD_HEADER_FMT,
        len(base_labels),
        len(merges),
        len(token_seq),
        bit_width,
        packed_size,
    )

    merges_flat: list[int] = []
    for left, right in merges:
        merges_flat.extend((int(left), int(right)))

    raw = bytearray()
    raw.extend(header)
    raw.extend(pack_u32_list(base_labels))
    raw.extend(pack_u32_list(merges_flat))
    raw.extend(packed_tokens)
    return rans_encode_framed_bytes(bytes(raw), workers=workers), {
        "base_symbol_count": len(base_labels),
        "merge_count": len(merges),
        "seed_merge_count": int(seed_applied),
        "shared_merge_count": int(seed_applied),
        "token_count": len(token_seq),
        "bit_width": bit_width,
    }


def decode_bpeans_sequence(payload: bytes, workers: int = 1) -> list[int]:
    raw = rans_decode_auto_bytes(payload, workers=workers)
    if len(raw) < BPE_PAYLOAD_HEADER_SIZE:
        raise ValueError("Invalid BPE payload (too short)")

    base_count, merge_count, token_count, bit_width, packed_size = struct.unpack(
        BPE_PAYLOAD_HEADER_FMT,
        raw[:BPE_PAYLOAD_HEADER_SIZE],
    )
    if not (1 <= bit_width <= 32):
        raise ValueError("Invalid BPE payload (bad bit width)")

    expected_packed_size = (token_count * bit_width + 7) // 8
    if packed_size != expected_packed_size:
        raise ValueError("Invalid BPE payload (packed size mismatch)")

    pos = BPE_PAYLOAD_HEADER_SIZE
    base_bytes = base_count * 4
    merge_bytes = merge_count * 8
    if len(raw) < pos + base_bytes + merge_bytes + packed_size:
        raise ValueError("Invalid BPE payload (truncated body)")

    base_labels = unpack_u32_list(raw[pos : pos + base_bytes]) if base_bytes else []
    pos += base_bytes

    merge_flat = unpack_u32_list(raw[pos : pos + merge_bytes]) if merge_bytes else []
    pos += merge_bytes
    if len(merge_flat) != merge_count * 2:
        raise ValueError("Invalid BPE payload (merge table length mismatch)")

    packed = raw[pos : pos + packed_size]
    pos += packed_size
    if pos != len(raw):
        raise ValueError("Invalid BPE payload (trailing bytes)")

    token_seq = bitunpack(packed, token_count, bit_width) if token_count else []
    vocab_size = base_count + merge_count
    if any(token >= vocab_size for token in token_seq):
        raise ValueError("Invalid BPE payload (token id out of range)")

    merges: list[tuple[int, int]] = []
    for i in range(0, len(merge_flat), 2):
        left = int(merge_flat[i])
        right = int(merge_flat[i + 1])
        sym_id = base_count + (i // 2)
        if left >= sym_id or right >= sym_id:
            raise ValueError("Invalid BPE payload (cyclic or forward merge reference)")
        merges.append((left, right))

    memo: dict[int, list[int]] = {}

    def expand(sym: int) -> list[int]:
        if sym < base_count:
            return [int(base_labels[sym])]
        cached = memo.get(sym)
        if cached is not None:
            return cached
        idx = sym - base_count
        if idx < 0 or idx >= len(merges):
            raise ValueError("Invalid BPE payload (merge symbol out of range)")
        left, right = merges[idx]
        expanded = expand(left) + expand(right)
        memo[sym] = expanded
        return expanded

    out: list[int] = []
    out_extend = out.extend
    for token in token_seq:
        out_extend(expand(int(token)))
    return out


def encode_bpeans_payload(
    level4: list[int],
    max_merges: int = BPE_MAX_MERGES,
    min_pair_count: int = BPE_MIN_PAIR_COUNT,
    workers: int = 1,
) -> bytes:
    payload, _ = encode_bpeans_sequence(
        level4,
        max_merges=max_merges,
        min_pair_count=min_pair_count,
        workers=workers,
    )
    return payload


def decode_bpeans_payload(payload: bytes, workers: int = 1) -> list[int]:
    return decode_bpeans_sequence(payload, workers=workers)


def _pack_global_bpe_manifest_payload(
    inner_payload: bytes,
    manifest_items: list[tuple[int, int]],
) -> bytes:
    raw = bytearray()
    raw.extend(struct.pack(BPE_GDICTM_PAYLOAD_HEADER_FMT, len(manifest_items), len(inner_payload)))
    for gid, label in manifest_items:
        raw.extend(struct.pack("<II", int(gid), int(label)))
    raw.extend(inner_payload)
    return bytes(raw)


def _decode_global_bpe_manifest_payload(
    payload: bytes,
    workers: int = 1,
) -> tuple[dict[int, int], list[int]]:
    if len(payload) < BPE_GDICTM_PAYLOAD_HEADER_SIZE:
        raise ValueError("Invalid BPE global-manifest payload (too short)")

    manifest_count, inner_size = struct.unpack(
        BPE_GDICTM_PAYLOAD_HEADER_FMT,
        payload[:BPE_GDICTM_PAYLOAD_HEADER_SIZE],
    )
    pos = BPE_GDICTM_PAYLOAD_HEADER_SIZE
    manifest_bytes = manifest_count * 8
    if len(payload) != pos + manifest_bytes + inner_size:
        raise ValueError("Invalid BPE global-manifest payload (size mismatch)")

    manifest: dict[int, int] = {}
    end_manifest = pos + manifest_bytes
    while pos < end_manifest:
        gid, label = struct.unpack("<II", payload[pos : pos + 8])
        gid = int(gid)
        if gid in manifest:
            raise ValueError("Invalid BPE global-manifest payload (duplicate gid)")
        manifest[gid] = int(label)
        pos += 8

    gid_seq = decode_bpeans_sequence(payload[pos : pos + inner_size], workers=workers)
    return manifest, gid_seq


def choose_best_payload(level4: list[int], workers: int = 1) -> tuple[int, bytes]:
    raw = pack_u32_list(level4)
    dict_pack = encode_dict_bitpack(level4)
    ans_raw = rans_encode_framed_bytes(raw, workers=workers)
    ans_dict = rans_encode_framed_bytes(dict_pack, workers=workers)
    ans_bpe = encode_bpeans_payload(level4, workers=workers)

    candidates = [
        (MODE_RAW_L4, raw),
        (MODE_ZLIB_RAW_L4, zlib.compress(raw, level=9)),
        (MODE_DICT_BITPACK, dict_pack),
        (MODE_ZLIB_DICT_BITPACK, zlib.compress(dict_pack, level=9)),
        (MODE_RANS_RAW_L4, ans_raw),
        (MODE_RANS_DICT_BITPACK, ans_dict),
        (MODE_RANS_BPE, ans_bpe),
    ]
    return min(candidates, key=lambda x: len(x[1]))


def encode_payload(level4: list[int], mode: str = "auto", workers: int = 1) -> tuple[int, bytes]:
    if mode == "auto":
        return choose_best_payload(level4, workers=workers)

    if mode not in MODE_NAME_TO_ID:
        raise ValueError(f"Unknown mode '{mode}'")

    mode_id = MODE_NAME_TO_ID[mode]
    raw = pack_u32_list(level4)
    dict_pack = encode_dict_bitpack(level4)

    if mode_id == MODE_RAW_L4:
        return mode_id, raw
    if mode_id == MODE_ZLIB_RAW_L4:
        return mode_id, zlib.compress(raw, level=9)
    if mode_id == MODE_DICT_BITPACK:
        return mode_id, dict_pack
    if mode_id == MODE_ZLIB_DICT_BITPACK:
        return mode_id, zlib.compress(dict_pack, level=9)
    if mode_id == MODE_RANS_RAW_L4:
        return mode_id, rans_encode_framed_bytes(raw, workers=workers)
    if mode_id == MODE_RANS_DICT_BITPACK:
        return mode_id, rans_encode_framed_bytes(dict_pack, workers=workers)
    if mode_id == MODE_RANS_BPE:
        return mode_id, encode_bpeans_payload(level4, workers=workers)
    if mode_id in (
        MODE_GLOBAL_DICT,
        MODE_ZGLOBAL_DICT,
        MODE_GLOBAL_DICT_MANIFEST,
        MODE_ZGLOBAL_DICT_MANIFEST,
        MODE_RANS_BPE_GLOBAL_DICT,
        MODE_RANS_BPE_GLOBAL_DICT_MANIFEST,
    ):
        raise ValueError("Global-dictionary BPE modes require file-based API")
    raise ValueError(f"Unsupported mode id {mode_id}")


def decode_payload(mode_id: int, payload: bytes, workers: int = 1) -> list[int]:
    if mode_id == MODE_RAW_L4:
        return unpack_u32_list(payload)
    if mode_id == MODE_ZLIB_RAW_L4:
        return unpack_u32_list(zlib.decompress(payload))
    if mode_id == MODE_DICT_BITPACK:
        return decode_dict_bitpack(payload)
    if mode_id == MODE_ZLIB_DICT_BITPACK:
        return decode_dict_bitpack(zlib.decompress(payload))
    if mode_id == MODE_RANS_RAW_L4:
        return unpack_u32_list(rans_decode_auto_bytes(payload, workers=workers))
    if mode_id == MODE_RANS_DICT_BITPACK:
        return decode_dict_bitpack(rans_decode_auto_bytes(payload, workers=workers))
    if mode_id == MODE_RANS_BPE:
        return decode_bpeans_payload(payload, workers=workers)
    if mode_id == MODE_GLOBAL_DICT_MANIFEST:
        return decode_global_manifest_bitpack(payload)
    if mode_id == MODE_ZGLOBAL_DICT_MANIFEST:
        return decode_global_manifest_bitpack(zlib.decompress(payload))
    if mode_id in (
        MODE_GLOBAL_DICT,
        MODE_ZGLOBAL_DICT,
        MODE_RANS_BPE_GLOBAL_DICT,
        MODE_RANS_BPE_GLOBAL_DICT_MANIFEST,
    ):
        raise ValueError("Global-dictionary modes require file-based API with --global-dict-db when applicable")
    raise ValueError(f"Unknown payload mode: {mode_id}")


def pack_header_v1(mode_id: int, original_size: int, payload_size: int, checksum: bytes) -> bytes:
    return struct.pack(
        HEADER_V1_FMT,
        MAGIC_V1,
        mode_id,
        original_size,
        payload_size,
        checksum,
    )


def pack_header_v2(
    mode_id: int,
    original_size: int,
    payload_size: int,
    metadata_size: int,
    checksum: bytes,
) -> bytes:
    return struct.pack(
        HEADER_V2_FMT,
        MAGIC_V2,
        mode_id,
        original_size,
        payload_size,
        metadata_size,
        checksum,
    )


def build_container_blob(
    mode_id: int,
    original_data: bytes,
    payload: bytes,
    metadata: dict | None = None,
) -> bytes:
    checksum = hashlib.blake2b(original_data, digest_size=16).digest()
    meta_blob = serialize_metadata(metadata)
    header = pack_header_v2(
        mode_id=mode_id,
        original_size=len(original_data),
        payload_size=len(payload),
        metadata_size=len(meta_blob),
        checksum=checksum,
    )
    return header + meta_blob + payload


def parse_container_blob(container_blob: bytes) -> tuple[int, int, bytes, bytes, dict, bytes, int]:
    if len(container_blob) < 8:
        raise ValueError("Invalid container: header too short")

    magic = container_blob[:8]
    if magic == MAGIC_V1:
        if len(container_blob) < HEADER_V1_SIZE:
            raise ValueError("Invalid container: v1 header too short")
        header = container_blob[:HEADER_V1_SIZE]
        _, mode_id, original_size, payload_size, checksum = struct.unpack(HEADER_V1_FMT, header)
        payload_offset = HEADER_V1_SIZE
        metadata = {}
    elif magic == MAGIC_V2:
        if len(container_blob) < HEADER_V2_SIZE:
            raise ValueError("Invalid container: v2 header too short")
        header = container_blob[:HEADER_V2_SIZE]
        _, mode_id, original_size, payload_size, meta_size, checksum = struct.unpack(HEADER_V2_FMT, header)
        payload_offset = HEADER_V2_SIZE + meta_size
        if len(container_blob) < payload_offset:
            raise ValueError("Invalid container: metadata truncated")
        metadata = deserialize_metadata(container_blob[HEADER_V2_SIZE:payload_offset])
    else:
        raise ValueError("Invalid container: bad magic")

    expected_size = payload_offset + payload_size
    if len(container_blob) != expected_size:
        if len(container_blob) < expected_size:
            raise ValueError("Invalid container: payload truncated")
        raise ValueError("Invalid container: trailing bytes")

    payload = container_blob[payload_offset:expected_size]
    return mode_id, original_size, checksum, payload, metadata, magic, payload_offset


def read_container_header(input_path: Path) -> tuple[int, int, int, bytes, dict, bytes, int]:
    total_size = input_path.stat().st_size
    with input_path.open("rb") as f:
        magic = f.read(8)
        if len(magic) != 8:
            raise ValueError("Invalid container: header too short")

        if magic == MAGIC_V1:
            rest = _read_exact(f, HEADER_V1_SIZE - 8)
            _, mode_id, original_size, payload_size, checksum = struct.unpack(
                HEADER_V1_FMT, magic + rest
            )
            metadata = {}
            payload_offset = HEADER_V1_SIZE
        elif magic == MAGIC_V2:
            rest = _read_exact(f, HEADER_V2_SIZE - 8)
            _, mode_id, original_size, payload_size, meta_size, checksum = struct.unpack(
                HEADER_V2_FMT, magic + rest
            )
            meta_blob = _read_exact(f, meta_size)
            metadata = deserialize_metadata(meta_blob)
            payload_offset = HEADER_V2_SIZE + meta_size
        else:
            raise ValueError("Invalid container: bad magic")

    if mode_id not in MODE_ID_TO_NAME:
        raise ValueError(f"Unknown payload mode: {mode_id}")

    expected_total = payload_offset + payload_size
    if total_size != expected_total:
        if total_size < expected_total:
            raise ValueError("Invalid container: payload truncated")
        raise ValueError("Invalid container: trailing bytes")
    return mode_id, original_size, payload_size, checksum, metadata, magic, payload_offset


def decode_container(
    mode_id: int,
    original_size: int,
    checksum: bytes,
    payload: bytes,
    workers: int = 1,
) -> bytes:
    level4 = decode_payload(mode_id, payload, workers=workers)
    expected_n4 = level_counts(original_size)[4]
    if len(level4) != expected_n4:
        raise ValueError(f"Corrupted container: expected {expected_n4} L4 labels, got {len(level4)}")

    data = restore_bytes_from_level4(level4, original_size)
    if len(data) != original_size:
        raise ValueError("Corrupted container: bad restored size")

    actual_checksum = hashlib.blake2b(data, digest_size=16).digest()
    if actual_checksum != checksum:
        raise ValueError("Checksum mismatch: decoded data is not identical to original")
    return data


def compress_bytes(
    data: bytes,
    mode: str = "auto",
    metadata: dict | None = None,
    workers: int = 1,
) -> tuple[bytes, dict]:
    level4 = build_level4_labels(data)
    mode_id, payload = encode_payload(level4, mode=mode, workers=workers)
    blob = build_container_blob(mode_id, data, payload, metadata=metadata)

    stats = {
        "input_size": len(data),
        "output_size": len(blob),
        "ratio": (len(data) / len(blob)) if len(blob) else 0.0,
        "mode_id": mode_id,
        "mode_name": mode_name(mode_id),
        "n4": len(level4),
        "streaming": False,
        "workers": max(1, int(workers)),
    }
    return blob, stats


def decompress_bytes(container_blob: bytes, workers: int = 1) -> tuple[bytes, dict]:
    mode_id, original_size, checksum, payload, metadata, _, _ = parse_container_blob(container_blob)
    data = decode_container(mode_id, original_size, checksum, payload, workers=workers)

    stats = {
        "output_size": len(data),
        "mode_id": mode_id,
        "mode_name": mode_name(mode_id),
        "streaming": False,
        "metadata": metadata,
        "workers": max(1, int(workers)),
    }
    return data, stats


class L4EncoderState:
    def __init__(self) -> None:
        self.carry_byte: int | None = None
        self.carry_l3: int | None = None

    def feed(self, chunk: bytes) -> array:
        out = array("I")
        carry_byte = self.carry_byte
        carry_l3 = self.carry_l3

        for b in chunk:
            if carry_byte is None:
                carry_byte = b
                continue
            l3 = (carry_byte << 8) | b
            carry_byte = None

            if carry_l3 is None:
                carry_l3 = l3
            else:
                out.append((carry_l3 << 16) | l3)
                carry_l3 = None

        self.carry_byte = carry_byte
        self.carry_l3 = carry_l3
        return out

    def finalize(self) -> array:
        out = array("I")
        carry_byte = self.carry_byte
        carry_l3 = self.carry_l3

        if carry_byte is not None:
            l3 = carry_byte << 8
            carry_byte = None
            if carry_l3 is None:
                carry_l3 = l3
            else:
                out.append((carry_l3 << 16) | l3)
                carry_l3 = None

        if carry_l3 is not None:
            out.append(carry_l3 << 16)
            carry_l3 = None

        self.carry_byte = carry_byte
        self.carry_l3 = carry_l3
        return out


def emit_decoded_l4_words(
    words: array,
    out_f,
    hasher,
    original_size: int,
    produced_bytes: int,
) -> int:
    decoded = bytearray()
    decoded_extend = decoded.extend
    for word in words:
        decoded_extend(
            (
                (word >> 24) & 0xFF,
                (word >> 16) & 0xFF,
                (word >> 8) & 0xFF,
                word & 0xFF,
            )
        )

    remaining = original_size - produced_bytes
    if remaining <= 0:
        return produced_bytes
    if len(decoded) > remaining:
        decoded = decoded[:remaining]
    if decoded:
        out_f.write(decoded)
        hasher.update(decoded)
        produced_bytes += len(decoded)
    return produced_bytes


class RawL4PayloadStreamingDecoder:
    def __init__(self, output_path: Path, original_size: int, expected_checksum: bytes):
        self.output_path = output_path
        self.original_size = original_size
        self.expected_checksum = expected_checksum
        self.expected_n4 = level_counts(original_size)[4]

        self._out_f = output_path.open("wb")
        self._hasher = hashlib.blake2b(digest_size=16)
        self._carry = b""
        self._produced_n4 = 0
        self._produced_bytes = 0
        self._closed = False

    def feed(self, data: bytes | memoryview) -> None:
        if self._closed:
            raise ValueError("Decoder already closed")
        if not data:
            return

        buf = self._carry + bytes(data)
        complete = (len(buf) // 4) * 4
        if complete:
            arr = array("I")
            arr.frombytes(buf[:complete])
            if sys.byteorder != "little":
                arr.byteswap()
            self._produced_n4 += len(arr)
            self._produced_bytes = emit_decoded_l4_words(
                arr,
                self._out_f,
                self._hasher,
                self.original_size,
                self._produced_bytes,
            )
        self._carry = buf[complete:]

    def finish(self) -> int:
        if self._closed:
            raise ValueError("Decoder already closed")
        try:
            if self._carry:
                raise ValueError("Corrupted RAW payload: trailing non-u32 bytes")
            if self._produced_n4 != self.expected_n4:
                raise ValueError(
                    f"Corrupted container: expected {self.expected_n4} L4 labels, got {self._produced_n4}"
                )
            if self._produced_bytes != self.original_size:
                raise ValueError(
                    f"Corrupted container: expected {self.original_size} bytes, got {self._produced_bytes}"
                )
            self._out_f.close()
            self._closed = True
            if self._hasher.digest() != self.expected_checksum:
                raise ValueError("Checksum mismatch: decoded data is not identical to original")
            return self._produced_bytes
        except Exception:
            self.abort()
            raise

    def abort(self) -> None:
        if not self._closed:
            try:
                self._out_f.close()
            finally:
                self._closed = True
            try:
                if self.output_path.exists():
                    self.output_path.unlink()
            except Exception:
                pass


class RansStreamingDecoder:
    def __init__(self, emit_bytes):
        self.emit_bytes = emit_bytes
        self._head_size = struct.calcsize("<QH")
        self._head_buf = bytearray()
        self._table_buf = bytearray()
        self._state_buf = bytearray()
        self._renorm_buf = bytearray()
        self._renorm_pos = 0

        self._header_stage = "head"
        self._raw_size = 0
        self._symbol_count = 0
        self._table_bytes_needed = 0
        self._state = 0
        self._produced = 0
        self._decode_table: bytearray | None = None
        self._freqs: dict[int, int] = {}
        self._starts: dict[int, int] = {}
        self._out_buf = bytearray()
        self._closed = False

    def _flush_out(self) -> None:
        if self._out_buf:
            self.emit_bytes(self._out_buf)
            self._out_buf.clear()

    def _parse_table(self) -> None:
        pos = 0
        decode_table = bytearray(RANS_TOTAL_FREQ)
        freqs: dict[int, int] = {}
        starts: dict[int, int] = {}
        cum = 0
        entry_size = struct.calcsize("<BH")
        for _ in range(self._symbol_count):
            sym, freq = struct.unpack("<BH", self._table_buf[pos : pos + entry_size])
            pos += entry_size
            if freq <= 0:
                raise ValueError("Invalid rANS payload (zero frequency)")
            if sym in freqs:
                raise ValueError("Invalid rANS payload (duplicate symbol)")
            freqs[int(sym)] = int(freq)
            starts[int(sym)] = cum
            decode_table[cum : cum + freq] = bytes([sym]) * freq
            cum += freq
        if cum != RANS_TOTAL_FREQ:
            raise ValueError("Invalid rANS payload (frequency total mismatch)")
        self._freqs = freqs
        self._starts = starts
        self._decode_table = decode_table

    def _decode_available(self) -> None:
        if self._decode_table is None:
            return

        while self._produced < self._raw_size:
            while self._state < RANS_L:
                if self._renorm_pos >= len(self._renorm_buf):
                    self._flush_out()
                    if self._renorm_pos:
                        del self._renorm_buf[: self._renorm_pos]
                        self._renorm_pos = 0
                    return
                self._state = (self._state << 8) | self._renorm_buf[self._renorm_pos]
                self._renorm_pos += 1

            slot = self._state & (RANS_TOTAL_FREQ - 1)
            sym = self._decode_table[slot]
            self._out_buf.append(sym)
            self._produced += 1
            self._state = self._freqs[sym] * (self._state >> RANS_SCALE_BITS) + (slot - self._starts[sym])

            if len(self._out_buf) >= (1 << 20):
                self._flush_out()

        while self._state < RANS_L and self._renorm_pos < len(self._renorm_buf):
            self._state = (self._state << 8) | self._renorm_buf[self._renorm_pos]
            self._renorm_pos += 1

        self._flush_out()

    def feed(self, data: bytes | memoryview) -> None:
        if self._closed:
            raise ValueError("Decoder already closed")
        if not data:
            return

        mv = memoryview(data)
        pos = 0
        mv_len = len(mv)

        while pos < mv_len:
            if self._header_stage == "done":
                raise ValueError("Invalid rANS payload (unexpected trailing bytes)")

            if self._header_stage == "head":
                need = self._head_size - len(self._head_buf)
                take = min(need, mv_len - pos)
                self._head_buf.extend(mv[pos : pos + take])
                pos += take
                if len(self._head_buf) == self._head_size:
                    self._raw_size, self._symbol_count = struct.unpack("<QH", self._head_buf)
                    if self._symbol_count == 0:
                        if self._raw_size != 0:
                            raise ValueError("Invalid rANS payload (empty table mismatch)")
                        self._header_stage = "done"
                        if pos != mv_len:
                            raise ValueError("Invalid rANS payload (unexpected bytes for empty stream)")
                        break
                    self._table_bytes_needed = self._symbol_count * struct.calcsize("<BH")
                    self._header_stage = "table"
                continue

            if self._header_stage == "table":
                need = self._table_bytes_needed - len(self._table_buf)
                take = min(need, mv_len - pos)
                self._table_buf.extend(mv[pos : pos + take])
                pos += take
                if len(self._table_buf) == self._table_bytes_needed:
                    self._parse_table()
                    self._header_stage = "state"
                continue

            if self._header_stage == "state":
                need = 4 - len(self._state_buf)
                take = min(need, mv_len - pos)
                self._state_buf.extend(mv[pos : pos + take])
                pos += take
                if len(self._state_buf) == 4:
                    self._state = struct.unpack("<I", self._state_buf)[0]
                    self._header_stage = "stream"
                    self._decode_available()
                continue

            if self._header_stage == "stream":
                self._renorm_buf.extend(mv[pos:])
                pos = mv_len
                self._decode_available()
                continue

            raise ValueError("Invalid rANS decoder state")

    def finish(self) -> None:
        if self._closed:
            raise ValueError("Decoder already closed")
        try:
            if self._header_stage == "head":
                raise ValueError("Invalid rANS payload (missing header)")
            if self._header_stage == "table":
                raise ValueError("Invalid rANS payload (truncated frequency table)")
            if self._header_stage == "state":
                raise ValueError("Invalid rANS payload (missing initial state)")
            if self._raw_size == 0:
                if self._header_stage != "done" or self._renorm_buf:
                    raise ValueError("Invalid rANS payload (unexpected bytes for empty stream)")
            else:
                self._decode_available()
                if self._produced != self._raw_size:
                    raise ValueError("Invalid rANS payload (truncated renormalization stream)")
                if self._renorm_pos != len(self._renorm_buf):
                    raise ValueError("Invalid rANS payload (trailing bytes)")
            self._closed = True
        finally:
            self._closed = True


class FramedRansStreamingDecoder:
    def __init__(self, emit_bytes):
        self.emit_bytes = emit_bytes
        self._head_buf = bytearray()
        self._block_head_buf = bytearray()
        self._block_buf = bytearray()
        self._header_done = False
        self._closed = False
        self._expected_block_size = 0
        self._current_raw_size = 0
        self._current_encoded_size = 0

    def feed(self, data: bytes | memoryview) -> None:
        if self._closed:
            raise ValueError("Decoder already closed")
        if not data:
            return

        mv = memoryview(data)
        pos = 0
        mv_len = len(mv)

        while pos < mv_len:
            if not self._header_done:
                need = RANS_FRAME_HEADER_SIZE - len(self._head_buf)
                take = min(need, mv_len - pos)
                self._head_buf.extend(mv[pos : pos + take])
                pos += take
                if len(self._head_buf) == RANS_FRAME_HEADER_SIZE:
                    magic, block_size = struct.unpack(RANS_FRAME_HEADER_FMT, self._head_buf)
                    if magic != RANS_FRAME_MAGIC:
                        raise ValueError("Invalid framed rANS payload (bad magic)")
                    if block_size <= 0:
                        raise ValueError("Invalid framed rANS payload (bad block size)")
                    self._expected_block_size = int(block_size)
                    self._header_done = True
                continue

            if self._current_encoded_size == 0:
                need = RANS_FRAME_BLOCK_SIZE - len(self._block_head_buf)
                take = min(need, mv_len - pos)
                self._block_head_buf.extend(mv[pos : pos + take])
                pos += take
                if len(self._block_head_buf) == RANS_FRAME_BLOCK_SIZE:
                    self._current_raw_size, self._current_encoded_size = struct.unpack(
                        RANS_FRAME_BLOCK_FMT,
                        self._block_head_buf,
                    )
                    self._block_head_buf.clear()
                continue

            need = self._current_encoded_size - len(self._block_buf)
            take = min(need, mv_len - pos)
            self._block_buf.extend(mv[pos : pos + take])
            pos += take
            if len(self._block_buf) == self._current_encoded_size:
                raw = rans_decode_bytes(bytes(self._block_buf))
                if len(raw) != self._current_raw_size:
                    raise ValueError("Invalid framed rANS payload (block size mismatch)")
                self.emit_bytes(raw)
                self._block_buf.clear()
                self._current_raw_size = 0
                self._current_encoded_size = 0

    def finish(self) -> None:
        if self._closed:
            raise ValueError("Decoder already closed")
        try:
            if len(self._head_buf) not in (0, RANS_FRAME_HEADER_SIZE):
                raise ValueError("Invalid framed rANS payload (truncated header)")
            if self._block_head_buf:
                raise ValueError("Invalid framed rANS payload (truncated block header)")
            if self._block_buf:
                raise ValueError("Invalid framed rANS payload (truncated block)")
            self._closed = True
        finally:
            self._closed = True


class AutoRansStreamingDecoder:
    def __init__(self, emit_bytes):
        self.emit_bytes = emit_bytes
        self._probe = bytearray()
        self._inner = None
        self._closed = False

    def _ensure_inner(self, final: bool = False) -> None:
        if self._inner is not None:
            return
        if not final and len(self._probe) < len(RANS_FRAME_MAGIC):
            return
        if self._probe.startswith(RANS_FRAME_MAGIC):
            self._inner = FramedRansStreamingDecoder(self.emit_bytes)
        else:
            self._inner = RansStreamingDecoder(self.emit_bytes)
        if self._probe:
            self._inner.feed(self._probe)
            self._probe.clear()

    def feed(self, data: bytes | memoryview) -> None:
        if self._closed:
            raise ValueError("Decoder already closed")
        if not data:
            return
        if self._inner is not None:
            self._inner.feed(data)
            return
        self._probe.extend(data)
        self._ensure_inner(final=False)

    def finish(self) -> None:
        if self._closed:
            raise ValueError("Decoder already closed")
        try:
            self._ensure_inner(final=True)
            if self._inner is None:
                raise ValueError("Invalid rANS payload (missing header)")
            self._inner.finish()
            self._closed = True
        finally:
            self._closed = True


def compress_file_stream_raw_or_zraw(
    input_path: Path,
    output_path: Path,
    mode_id: int,
    chunk_size: int,
    workers: int = 1,
    metadata: dict | None = None,
    verbose: bool = True,
) -> dict:
    if mode_id not in (MODE_RAW_L4, MODE_ZLIB_RAW_L4, MODE_RANS_RAW_L4):
        raise ValueError("Streaming encoder supports only raw/zraw/ansraw modes")

    original_size = input_path.stat().st_size
    meta_blob = serialize_metadata(metadata)
    expected_n4 = level_counts(original_size)[4]
    hasher = hashlib.blake2b(digest_size=16)
    l4_count = 0
    payload_size = 0

    workers = max(1, int(workers))
    if mode_id == MODE_RANS_RAW_L4:
        with input_path.open("rb") as in_f, output_path.open("wb") as out_f:
            out_f.write(b"\x00" * (HEADER_V2_SIZE + len(meta_blob)))
            encoder = FramedRansEncoder(out_f.write, block_size=max(1 << 16, chunk_size))

            for raw_chunk, words_cnt in _iter_l4_raw_chunks(
                input_stream=in_f,
                chunk_size=chunk_size,
                workers=workers,
                hasher=hasher,
            ):
                l4_count += words_cnt
                encoder.feed(raw_chunk)

            payload_size = encoder.finish()

            if l4_count != expected_n4:
                raise ValueError(f"Internal error: L4 count mismatch {l4_count} != {expected_n4}")

            checksum = hasher.digest()
            header = pack_header_v2(
                mode_id=mode_id,
                original_size=original_size,
                payload_size=payload_size,
                metadata_size=len(meta_blob),
                checksum=checksum,
            )
            out_f.seek(0)
            out_f.write(header)
            if meta_blob:
                out_f.write(meta_blob)
    else:
        compressor = zlib.compressobj(level=9) if mode_id == MODE_ZLIB_RAW_L4 else None

        with input_path.open("rb") as in_f, output_path.open("wb") as out_f:
            out_f.write(b"\x00" * (HEADER_V2_SIZE + len(meta_blob)))

            def emit_raw(raw_bytes: bytes) -> None:
                nonlocal payload_size
                if not raw_bytes:
                    return
                if compressor is not None:
                    comp = compressor.compress(raw_bytes)
                    if comp:
                        out_f.write(comp)
                        payload_size += len(comp)
                else:
                    out_f.write(raw_bytes)
                    payload_size += len(raw_bytes)

            for raw_chunk, words_cnt in _iter_l4_raw_chunks(
                input_stream=in_f,
                chunk_size=chunk_size,
                workers=workers,
                hasher=hasher,
            ):
                l4_count += words_cnt
                emit_raw(raw_chunk)

            if compressor is not None:
                comp = compressor.flush()
                if comp:
                    out_f.write(comp)
                    payload_size += len(comp)

            if l4_count != expected_n4:
                raise ValueError(f"Internal error: L4 count mismatch {l4_count} != {expected_n4}")

            checksum = hasher.digest()
            header = pack_header_v2(
                mode_id=mode_id,
                original_size=original_size,
                payload_size=payload_size,
                metadata_size=len(meta_blob),
                checksum=checksum,
            )
            out_f.seek(0)
            out_f.write(header)
            if meta_blob:
                out_f.write(meta_blob)

    out_size = output_path.stat().st_size
    stats = {
        "input_size": original_size,
        "output_size": out_size,
        "ratio": (original_size / out_size) if out_size else 0.0,
        "mode_id": mode_id,
        "mode_name": mode_name(mode_id),
        "n4": l4_count,
        "streaming": True,
    }

    if verbose:
        n0, n1, n2, n3, n4 = level_counts(original_size)
        print(f"Input:  {input_path} ({original_size} bytes)")
        print(
            f"Output: {output_path} ({out_size} bytes), "
            f"mode={stats['mode_name']}, ratio={stats['ratio']:.3f}x, stream=yes"
        )
        print(f"Levels: L0={n0}, L1={n1}, L2={n2}, L3={n3}, L4={n4}")
    return stats


def decompress_file_stream_raw_or_zraw(
    input_path: Path,
    output_path: Path,
    mode_id: int,
    original_size: int,
    payload_size: int,
    checksum: bytes,
    payload_offset: int,
    chunk_size: int,
    verbose: bool = True,
) -> dict:
    if mode_id not in (MODE_RAW_L4, MODE_ZLIB_RAW_L4, MODE_RANS_RAW_L4):
        raise ValueError("Streaming decoder supports only raw/zraw/ansraw modes")

    if mode_id == MODE_RANS_RAW_L4:
        raw_decoder = RawL4PayloadStreamingDecoder(
            output_path=output_path,
            original_size=original_size,
            expected_checksum=checksum,
        )
        rans_decoder = AutoRansStreamingDecoder(raw_decoder.feed)

        try:
            with input_path.open("rb") as in_f:
                in_f.seek(payload_offset)
                remaining = payload_size
                while remaining > 0:
                    comp_chunk = in_f.read(min(chunk_size, remaining))
                    if not comp_chunk:
                        raise ValueError("Invalid container: payload truncated")
                    remaining -= len(comp_chunk)
                    rans_decoder.feed(comp_chunk)
            rans_decoder.finish()
            produced_bytes = raw_decoder.finish()
        except Exception:
            raw_decoder.abort()
            raise

        stats = {
            "output_size": produced_bytes,
            "mode_id": mode_id,
            "mode_name": mode_name(mode_id),
            "streaming": True,
        }
        if verbose:
            print(f"Decompressed: {output_path} ({produced_bytes} bytes), mode={stats['mode_name']}, stream=yes")
        return stats

    expected_n4 = level_counts(original_size)[4]
    produced_bytes = 0
    produced_n4 = 0
    hasher = hashlib.blake2b(digest_size=16)

    with input_path.open("rb") as in_f, output_path.open("wb") as out_f:
        in_f.seek(payload_offset)

        if mode_id == MODE_RAW_L4:
            expected_payload_size = expected_n4 * 4
            if payload_size != expected_payload_size:
                raise ValueError(
                    f"Corrupted container: RAW payload size mismatch {payload_size} != {expected_payload_size}"
                )

            carry = b""
            remaining = payload_size
            while remaining > 0:
                chunk = in_f.read(min(chunk_size, remaining))
                if not chunk:
                    raise ValueError("Invalid container: payload truncated")
                remaining -= len(chunk)

                buf = carry + chunk
                complete = (len(buf) // 4) * 4
                if complete:
                    arr = array("I")
                    arr.frombytes(buf[:complete])
                    if sys.byteorder != "little":
                        arr.byteswap()
                    produced_n4 += len(arr)
                    produced_bytes = emit_decoded_l4_words(arr, out_f, hasher, original_size, produced_bytes)
                carry = buf[complete:]

            if carry:
                raise ValueError("Corrupted RAW payload: trailing non-u32 bytes")

        else:
            decomp = zlib.decompressobj()
            carry = b""
            remaining = payload_size

            while remaining > 0:
                comp_chunk = in_f.read(min(chunk_size, remaining))
                if not comp_chunk:
                    raise ValueError("Invalid container: payload truncated")
                remaining -= len(comp_chunk)

                dec = decomp.decompress(comp_chunk)
                if dec:
                    buf = carry + dec
                    complete = (len(buf) // 4) * 4
                    if complete:
                        arr = array("I")
                        arr.frombytes(buf[:complete])
                        if sys.byteorder != "little":
                            arr.byteswap()
                        produced_n4 += len(arr)
                        produced_bytes = emit_decoded_l4_words(arr, out_f, hasher, original_size, produced_bytes)
                    carry = buf[complete:]

            tail = decomp.flush()
            if tail:
                buf = carry + tail
            else:
                buf = carry

            complete = (len(buf) // 4) * 4
            if complete:
                arr = array("I")
                arr.frombytes(buf[:complete])
                if sys.byteorder != "little":
                    arr.byteswap()
                produced_n4 += len(arr)
                produced_bytes = emit_decoded_l4_words(arr, out_f, hasher, original_size, produced_bytes)
            carry = buf[complete:]
            if carry:
                raise ValueError("Corrupted zraw payload: trailing non-u32 bytes after inflate")

    if produced_n4 != expected_n4:
        raise ValueError(f"Corrupted container: expected {expected_n4} L4 labels, got {produced_n4}")
    if produced_bytes != original_size:
        raise ValueError(f"Corrupted container: expected {original_size} bytes, got {produced_bytes}")
    if hasher.digest() != checksum:
        raise ValueError("Checksum mismatch: decoded data is not identical to original")

    stats = {
        "output_size": produced_bytes,
        "mode_id": mode_id,
        "mode_name": mode_name(mode_id),
        "streaming": True,
    }
    if verbose:
        print(f"Decompressed: {output_path} ({produced_bytes} bytes), mode={stats['mode_name']}, stream=yes")
    return stats


def _write_sqlite_dictionary_blob(conn: sqlite3.Connection, write_bytes, batch_size: int = 1 << 16) -> int:
    total = 0
    arr = array("I")
    cur = conn.execute("SELECT label FROM dict_map ORDER BY idx")
    for (label,) in cur:
        arr.append(int(label))
        if len(arr) >= batch_size:
            raw = pack_u32_array(arr)
            write_bytes(raw)
            total += len(raw)
            arr = array("I")
    if arr:
        raw = pack_u32_array(arr)
        write_bytes(raw)
        total += len(raw)
    return total


def _assign_idx_sqlite(
    conn: sqlite3.Connection,
    cache: dict[int, int],
    label: int,
    next_idx: int,
    cache_limit: int = 1 << 18,
) -> tuple[int, int]:
    idx = cache.get(label)
    if idx is not None:
        return idx, next_idx

    row = conn.execute("SELECT idx FROM dict_map WHERE label=?", (label,)).fetchone()
    if row is None:
        idx = next_idx
        conn.execute("INSERT INTO dict_map(label, idx) VALUES (?, ?)", (label, idx))
        next_idx += 1
    else:
        idx = int(row[0])

    if len(cache) >= cache_limit:
        cache.clear()
    cache[label] = idx
    return idx, next_idx


def _open_global_dict_sqlite(db_path: Path, readonly: bool = False) -> sqlite3.Connection:
    if readonly:
        if not db_path.exists():
            raise ValueError(f"Global dictionary DB not found: {db_path}")
        uri = f"file:{db_path.resolve().as_posix()}?mode=ro"
        conn = sqlite3.connect(uri, uri=True)
        conn.execute("PRAGMA query_only=ON")
        return conn

    db_path.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(db_path)
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA synchronous=NORMAL")
    conn.execute("PRAGMA temp_store=MEMORY")
    conn.execute(
        "CREATE TABLE IF NOT EXISTS global_meta("
        "key TEXT PRIMARY KEY, "
        "value TEXT NOT NULL)"
    )
    conn.execute(
        "CREATE TABLE IF NOT EXISTS global_dict("
        "gid INTEGER PRIMARY KEY, "
        "label INTEGER NOT NULL UNIQUE)"
    )
    conn.execute(
        "CREATE TABLE IF NOT EXISTS global_snapshots("
        "snapshot_id INTEGER PRIMARY KEY AUTOINCREMENT, "
        "max_gid INTEGER NOT NULL, "
        "entry_count INTEGER NOT NULL, "
        "dict_hash TEXT NOT NULL, "
        "created_utc TEXT NOT NULL)"
    )
    conn.execute(
        "CREATE TABLE IF NOT EXISTS global_bpe_pairs("
        "left_gid INTEGER NOT NULL, "
        "right_gid INTEGER NOT NULL, "
        "score INTEGER NOT NULL DEFAULT 0, "
        "file_hits INTEGER NOT NULL DEFAULT 0, "
        "last_seen_utc TEXT NOT NULL DEFAULT '', "
        "PRIMARY KEY(left_gid, right_gid))"
    )
    conn.execute(
        "CREATE TABLE IF NOT EXISTS global_bpe_vocab("
        "rank INTEGER PRIMARY KEY, "
        "left_gid INTEGER NOT NULL, "
        "right_gid INTEGER NOT NULL, "
        "score INTEGER NOT NULL DEFAULT 0, "
        "file_hits INTEGER NOT NULL DEFAULT 0, "
        "last_seen_utc TEXT NOT NULL DEFAULT '', "
        "UNIQUE(left_gid, right_gid))"
    )
    conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_global_dict_label ON global_dict(label)")
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_global_bpe_pairs_score "
        "ON global_bpe_pairs(score DESC, file_hits DESC)"
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_global_bpe_vocab_score "
        "ON global_bpe_vocab(score DESC, file_hits DESC, rank ASC)"
    )
    if conn.execute("SELECT value FROM global_meta WHERE key='db_uuid'").fetchone() is None:
        conn.execute(
            "INSERT INTO global_meta(key, value) VALUES ('db_uuid', ?)",
            (uuid.uuid4().hex,),
        )
    if conn.execute("SELECT value FROM global_meta WHERE key='schema_version'").fetchone() is None:
        conn.execute(
            "INSERT INTO global_meta(key, value) VALUES ('schema_version', ?)",
            (str(GLOBAL_DICT_SCHEMA_VERSION),),
        )
    conn.commit()
    return conn


def _global_meta_get(conn: sqlite3.Connection, key: str, default: str | None = None) -> str | None:
    row = conn.execute("SELECT value FROM global_meta WHERE key=?", (key,)).fetchone()
    if row is None:
        return default
    return str(row[0])


def _global_meta_set(conn: sqlite3.Connection, key: str, value: str) -> None:
    conn.execute(
        "INSERT INTO global_meta(key, value) VALUES (?, ?) "
        "ON CONFLICT(key) DO UPDATE SET value=excluded.value",
        (key, value),
    )


def _compute_global_dict_state(conn: sqlite3.Connection, max_gid: int) -> tuple[int, str]:
    if max_gid < 0:
        raise ValueError("Invalid max_gid for global dictionary state")

    hasher = hashlib.blake2b(digest_size=16)
    entry_count = 0
    cur = conn.execute(
        "SELECT gid, label FROM global_dict WHERE gid <= ? ORDER BY gid",
        (int(max_gid),),
    )
    for gid, label in cur:
        hasher.update(struct.pack("<II", int(gid), int(label)))
        entry_count += 1

    hasher.update(struct.pack("<I", int(max_gid)))
    hasher.update(struct.pack("<I", int(entry_count)))
    return entry_count, hasher.hexdigest()


def _create_global_dict_snapshot(
    conn: sqlite3.Connection,
    max_gid: int,
) -> dict:
    db_uuid = _global_meta_get(conn, "db_uuid")
    schema_version = int(_global_meta_get(conn, "schema_version", str(GLOBAL_DICT_SCHEMA_VERSION)) or GLOBAL_DICT_SCHEMA_VERSION)
    if db_uuid is None:
        raise ValueError("Global dictionary DB is missing db_uuid metadata")

    entry_count, dict_hash = _compute_global_dict_state(conn, max_gid)
    created_utc = datetime.now(timezone.utc).isoformat()
    cur = conn.execute(
        "INSERT INTO global_snapshots(max_gid, entry_count, dict_hash, created_utc) VALUES (?, ?, ?, ?)",
        (int(max_gid), int(entry_count), dict_hash, created_utc),
    )
    snapshot_id = int(cur.lastrowid)
    _global_meta_set(conn, "last_snapshot_id", str(snapshot_id))
    conn.commit()

    return {
        "db_uuid": db_uuid,
        "schema_version": schema_version,
        "snapshot_id": snapshot_id,
        "max_gid": int(max_gid),
        "entry_count": int(entry_count),
        "dict_hash": dict_hash,
        "created_utc": created_utc,
    }


def _extract_container_global_dict_meta(metadata: dict | None) -> dict | None:
    if not metadata:
        return None
    raw = metadata.get("global_dict")
    if raw is None:
        return None
    if not isinstance(raw, dict):
        raise ValueError("Invalid container metadata: global_dict must be an object")

    required = ("db_uuid", "schema_version", "snapshot_id", "max_gid", "entry_count", "dict_hash")
    for key in required:
        if key not in raw:
            raise ValueError(f"Invalid container metadata: missing global_dict.{key}")

    return {
        "db_uuid": str(raw["db_uuid"]),
        "schema_version": int(raw["schema_version"]),
        "snapshot_id": int(raw["snapshot_id"]),
        "max_gid": int(raw["max_gid"]),
        "entry_count": int(raw["entry_count"]),
        "dict_hash": str(raw["dict_hash"]),
        "created_utc": str(raw.get("created_utc", "")),
        "embedded_manifest": bool(raw.get("embedded_manifest", False)),
        "manifest_count": int(raw.get("manifest_count", 0)),
    }


def _validate_global_dict_snapshot(conn: sqlite3.Connection, container_meta: dict | None) -> dict | None:
    if container_meta is None:
        return None

    db_uuid = _global_meta_get(conn, "db_uuid")
    schema_version = int(_global_meta_get(conn, "schema_version", str(GLOBAL_DICT_SCHEMA_VERSION)) or GLOBAL_DICT_SCHEMA_VERSION)
    if db_uuid != container_meta["db_uuid"]:
        raise ValueError(
            "Global dictionary DB UUID mismatch: container was encoded with a different dictionary"
        )
    if schema_version != container_meta["schema_version"]:
        raise ValueError(
            "Global dictionary schema version mismatch between container and SQLite DB"
        )

    row = conn.execute(
        "SELECT max_gid, entry_count, dict_hash FROM global_snapshots WHERE snapshot_id=?",
        (container_meta["snapshot_id"],),
    ).fetchone()
    if row is not None:
        snap_max_gid, snap_entry_count, snap_dict_hash = int(row[0]), int(row[1]), str(row[2])
        if (
            snap_max_gid != container_meta["max_gid"]
            or snap_entry_count != container_meta["entry_count"]
            or snap_dict_hash != container_meta["dict_hash"]
        ):
            raise ValueError(
                "Global dictionary snapshot metadata mismatch: container snapshot does not match DB snapshot record"
            )

    actual_entry_count, actual_hash = _compute_global_dict_state(conn, container_meta["max_gid"])
    if actual_entry_count != container_meta["entry_count"] or actual_hash != container_meta["dict_hash"]:
        raise ValueError(
            "Global dictionary snapshot mismatch: DB state differs from the snapshot required by this container"
        )
    return container_meta


def _assign_global_gid_sqlite(
    conn: sqlite3.Connection,
    cache: dict[int, int],
    label: int,
    cache_limit: int = 1 << 18,
) -> int:
    gid = cache.get(label)
    if gid is not None:
        return gid

    row = conn.execute("SELECT gid FROM global_dict WHERE label=?", (label,)).fetchone()
    if row is None:
        conn.execute("INSERT OR IGNORE INTO global_dict(label) VALUES (?)", (label,))
        row = conn.execute("SELECT gid FROM global_dict WHERE label=?", (label,)).fetchone()
        if row is None:
            raise ValueError("Failed to assign global dictionary id")
    gid = int(row[0])

    if gid < 0 or gid > 0xFFFFFFFF:
        raise ValueError("Global dictionary id overflow: only u32 ids are supported")

    if len(cache) >= cache_limit:
        cache.clear()
    cache[label] = gid
    return gid


def _mark_used_gid_sqlite(
    conn: sqlite3.Connection,
    cache: set[int],
    gid: int,
    cache_limit: int = 1 << 18,
) -> None:
    if gid in cache:
        return
    conn.execute("INSERT OR IGNORE INTO used_gid(gid) VALUES (?)", (int(gid),))
    if len(cache) >= cache_limit:
        cache.clear()
    cache.add(int(gid))


def _write_sqlite_global_manifest_blob(
    conn: sqlite3.Connection,
    write_bytes,
    batch_size: int = 1 << 15,
) -> tuple[int, int]:
    total = 0
    manifest_count = 0
    out = bytearray()
    cur = conn.execute(
        "SELECT g.gid, g.label "
        "FROM used_gid u "
        "JOIN global_dict g ON g.gid = u.gid "
        "ORDER BY g.gid"
    )
    for gid, label in cur:
        out.extend(struct.pack("<II", int(gid), int(label)))
        manifest_count += 1
        if manifest_count % batch_size == 0:
            write_bytes(out)
            total += len(out)
            out.clear()
    if out:
        write_bytes(out)
        total += len(out)
    return total, manifest_count


def _collect_sqlite_global_manifest_items(conn: sqlite3.Connection) -> list[tuple[int, int]]:
    cur = conn.execute(
        "SELECT g.gid, g.label "
        "FROM used_gid u "
        "JOIN global_dict g ON g.gid = u.gid "
        "ORDER BY g.gid"
    )
    return [(int(gid), int(label)) for gid, label in cur]


def _ensure_global_bpe_vocabulary(
    conn: sqlite3.Connection,
    vocab_limit: int = GLOBAL_BPE_VOCAB_LIMIT,
) -> int:
    existing = int(conn.execute("SELECT COUNT(*) FROM global_bpe_vocab").fetchone()[0])
    if existing > 0:
        return existing

    cur = conn.execute(
        "SELECT left_gid, right_gid, score, file_hits, last_seen_utc "
        "FROM global_bpe_pairs "
        "ORDER BY score DESC, file_hits DESC, left_gid ASC, right_gid ASC "
        "LIMIT ?",
        (int(vocab_limit),),
    )
    ranked = [
        (
            rank,
            int(left_gid),
            int(right_gid),
            int(score),
            int(file_hits),
            str(last_seen_utc),
        )
        for rank, (left_gid, right_gid, score, file_hits, last_seen_utc) in enumerate(cur, start=1)
    ]
    if ranked:
        conn.executemany(
            "INSERT INTO global_bpe_vocab(rank, left_gid, right_gid, score, file_hits, last_seen_utc) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            ranked,
        )
    return len(ranked)


def _load_global_bpe_seed_pairs(
    conn: sqlite3.Connection,
    limit: int = GLOBAL_BPE_SEED_LIMIT,
) -> list[tuple[int, int]]:
    _ensure_global_bpe_vocabulary(conn)
    cur = conn.execute(
        "SELECT left_gid, right_gid "
        "FROM global_bpe_vocab "
        "ORDER BY rank ASC "
        "LIMIT ?",
        (int(limit),),
    )
    return [(int(left), int(right)) for left, right in cur]


def _rebalance_global_bpe_vocabulary(
    conn: sqlite3.Connection,
    vocab_limit: int = GLOBAL_BPE_VOCAB_LIMIT,
) -> int:
    cur = conn.execute(
        "SELECT left_gid, right_gid, score, file_hits, last_seen_utc "
        "FROM global_bpe_vocab "
        "ORDER BY score DESC, file_hits DESC, last_seen_utc DESC, left_gid ASC, right_gid ASC "
        "LIMIT ?",
        (int(vocab_limit),),
    )
    ranked = [
        (
            rank,
            int(left_gid),
            int(right_gid),
            int(score),
            int(file_hits),
            str(last_seen_utc),
        )
        for rank, (left_gid, right_gid, score, file_hits, last_seen_utc) in enumerate(cur, start=1)
    ]
    conn.execute("DELETE FROM global_bpe_vocab")
    if ranked:
        conn.executemany(
            "INSERT INTO global_bpe_vocab(rank, left_gid, right_gid, score, file_hits, last_seen_utc) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            ranked,
        )
    return len(ranked)


def _update_global_bpe_pairs(
    conn: sqlite3.Connection,
    gid_seq: list[int],
    learn_limit: int = GLOBAL_BPE_LEARN_LIMIT,
    min_pair_count: int = BPE_MIN_PAIR_COUNT,
    pair_limit: int = GLOBAL_BPE_PAIR_LIMIT,
    vocab_limit: int = GLOBAL_BPE_VOCAB_LIMIT,
) -> dict:
    _ensure_global_bpe_vocabulary(conn, vocab_limit=vocab_limit)
    pair_counts = _count_adjacent_pairs(gid_seq)
    ranked = [
        (pair, count)
        for pair, count in pair_counts.items()
        if count >= min_pair_count
    ]
    ranked.sort(key=lambda item: (-item[1], item[0]))
    selected = ranked[: max(1, int(learn_limit))]
    if not selected:
        return {
            "updated_pairs": 0,
            "known_pairs": int(conn.execute("SELECT COUNT(*) FROM global_bpe_pairs").fetchone()[0]),
            "shared_vocab_size": int(conn.execute("SELECT COUNT(*) FROM global_bpe_vocab").fetchone()[0]),
        }

    now_utc = datetime.now(timezone.utc).isoformat()
    next_rank = int(conn.execute("SELECT COALESCE(MAX(rank), 0) FROM global_bpe_vocab").fetchone()[0]) + 1
    for (left_gid, right_gid), count in selected:
        conn.execute(
            "INSERT INTO global_bpe_pairs(left_gid, right_gid, score, file_hits, last_seen_utc) "
            "VALUES (?, ?, ?, 1, ?) "
            "ON CONFLICT(left_gid, right_gid) DO UPDATE SET "
            "score = global_bpe_pairs.score + excluded.score, "
            "file_hits = global_bpe_pairs.file_hits + 1, "
            "last_seen_utc = excluded.last_seen_utc",
            (int(left_gid), int(right_gid), int(count), now_utc),
        )
        conn.execute(
            "INSERT INTO global_bpe_vocab(rank, left_gid, right_gid, score, file_hits, last_seen_utc) "
            "VALUES (?, ?, ?, ?, 1, ?) "
            "ON CONFLICT(left_gid, right_gid) DO UPDATE SET "
            "score = global_bpe_vocab.score + excluded.score, "
            "file_hits = global_bpe_vocab.file_hits + 1, "
            "last_seen_utc = excluded.last_seen_utc",
            (int(next_rank), int(left_gid), int(right_gid), int(count), now_utc),
        )
        next_rank += 1

    total_pairs = int(conn.execute("SELECT COUNT(*) FROM global_bpe_pairs").fetchone()[0])
    if total_pairs > pair_limit:
        excess = total_pairs - pair_limit
        conn.execute(
            "DELETE FROM global_bpe_pairs "
            "WHERE rowid IN ("
            "SELECT rowid FROM global_bpe_pairs "
            "ORDER BY score ASC, file_hits ASC, last_seen_utc ASC, left_gid ASC, right_gid ASC "
            "LIMIT ?)",
            (int(excess),),
        )
        total_pairs = int(conn.execute("SELECT COUNT(*) FROM global_bpe_pairs").fetchone()[0])

    vocab_size = _rebalance_global_bpe_vocabulary(conn, vocab_limit=vocab_limit)
    return {
        "updated_pairs": len(selected),
        "known_pairs": total_pairs,
        "shared_vocab_size": vocab_size,
    }


def _read_u32_file_to_list(path: Path, chunk_words: int = 1 << 18) -> list[int]:
    out: list[int] = []
    with path.open("rb") as f:
        while True:
            raw = f.read(chunk_words * 4)
            if not raw:
                break
            if len(raw) % 4 != 0:
                raise ValueError("Corrupted temporary u32 stream")
            arr = array("I")
            arr.frombytes(raw)
            if sys.byteorder != "little":
                arr.byteswap()
            out.extend(int(v) for v in arr)
    return out


def _build_global_gid_sequence(
    conn: sqlite3.Connection,
    level4: list[int],
    collect_manifest: bool = False,
) -> tuple[list[int], int, list[tuple[int, int]] | None]:
    cache: dict[int, int] = {}
    gid_seq: list[int] = []
    max_gid = 0
    manifest_map: dict[int, int] | None = {} if collect_manifest else None

    for label in level4:
        gid = _assign_global_gid_sqlite(conn, cache, int(label))
        gid_seq.append(gid)
        if gid > max_gid:
            max_gid = gid
        if manifest_map is not None and gid not in manifest_map:
            manifest_map[gid] = int(label)

    manifest_items = None
    if manifest_map is not None:
        manifest_items = sorted(manifest_map.items())
    return gid_seq, max_gid, manifest_items


def compress_file_bpe_gdict_sqlite(
    input_path: Path,
    output_path: Path,
    mode_id: int,
    global_dict_db: Path | None,
    workers: int = 1,
    metadata: dict | None = None,
    verbose: bool = True,
) -> dict:
    if mode_id not in (MODE_RANS_BPE_GLOBAL_DICT, MODE_RANS_BPE_GLOBAL_DICT_MANIFEST):
        raise ValueError("BPE global-dictionary encoder supports only ansbpegdict/ansbpegdictm modes")
    if global_dict_db is None:
        raise ValueError("ansbpegdict/ansbpegdictm modes require --global-dict-db for encoding")

    data = input_path.read_bytes()
    original_size = len(data)
    level4 = build_level4_labels(data)
    expected_n4 = level_counts(original_size)[4]
    if len(level4) != expected_n4:
        raise ValueError(f"Internal error: L4 count mismatch {len(level4)} != {expected_n4}")

    db_path = Path(global_dict_db)
    conn = _open_global_dict_sqlite(db_path, readonly=False)
    try:
        if mode_id == MODE_RANS_BPE_GLOBAL_DICT_MANIFEST:
            conn.execute("CREATE TEMP TABLE IF NOT EXISTS used_gid(gid INTEGER PRIMARY KEY)")
            conn.execute("DELETE FROM used_gid")
            conn.commit()
        conn.execute("BEGIN")
        gid_seq, max_gid, manifest_items = _build_global_gid_sequence(
            conn,
            level4,
            collect_manifest=(mode_id == MODE_RANS_BPE_GLOBAL_DICT_MANIFEST),
        )
        seed_pairs = _load_global_bpe_seed_pairs(conn)
        bpe_cache_stats = _update_global_bpe_pairs(conn, gid_seq)
        conn.commit()
        snapshot_meta = _create_global_dict_snapshot(conn, max_gid=max_gid)
    finally:
        conn.close()

    container_metadata = dict(metadata or {})
    global_dict_meta = dict(snapshot_meta)
    if manifest_items is not None:
        global_dict_meta["embedded_manifest"] = True
        global_dict_meta["manifest_count"] = len(manifest_items)
    container_metadata["global_dict"] = global_dict_meta
    container_metadata["bpe"] = {
        "persistent_cache": True,
        "persistent_vocabulary": True,
        "seed_pairs_available": len(seed_pairs),
        "shared_vocab_pairs_available": len(seed_pairs),
    }

    if mode_id == MODE_RANS_BPE_GLOBAL_DICT:
        payload, bpe_stats = encode_bpeans_sequence(gid_seq, seed_pairs=seed_pairs, workers=workers)
    else:
        inner_payload, bpe_stats = encode_bpeans_sequence(gid_seq, seed_pairs=seed_pairs, workers=workers)
        payload = _pack_global_bpe_manifest_payload(inner_payload, manifest_items or [])

    container_metadata["bpe"]["seed_merges_applied"] = int(bpe_stats["seed_merge_count"])
    container_metadata["bpe"]["shared_vocab_merges_applied"] = int(bpe_stats["shared_merge_count"])
    container_metadata["bpe"]["merge_count"] = int(bpe_stats["merge_count"])
    container_metadata["bpe"]["known_pairs"] = int(bpe_cache_stats["known_pairs"])
    container_metadata["bpe"]["updated_pairs"] = int(bpe_cache_stats["updated_pairs"])
    container_metadata["bpe"]["shared_vocab_size"] = int(bpe_cache_stats["shared_vocab_size"])

    blob = build_container_blob(mode_id, data, payload, metadata=container_metadata)
    output_path.write_bytes(blob)

    out_size = len(blob)
    stats = {
        "input_size": original_size,
        "output_size": out_size,
        "ratio": (original_size / out_size) if out_size else 0.0,
        "mode_id": mode_id,
        "mode_name": mode_name(mode_id),
        "n4": len(level4),
        "streaming": False,
        "workers": max(1, int(workers)),
        "global_dict_db": str(db_path),
        "global_dict_snapshot_id": int(snapshot_meta["snapshot_id"]),
        "global_dict_db_uuid": str(snapshot_meta["db_uuid"]),
        "global_dict_max_gid": int(snapshot_meta["max_gid"]),
        "global_dict_hash": str(snapshot_meta["dict_hash"]),
        "bpe_seed_pairs_available": len(seed_pairs),
        "bpe_seed_merges_applied": int(bpe_stats["seed_merge_count"]),
        "bpe_shared_vocab_pairs_available": len(seed_pairs),
        "bpe_shared_vocab_merges_applied": int(bpe_stats["shared_merge_count"]),
        "bpe_merge_count": int(bpe_stats["merge_count"]),
        "bpe_known_pairs": int(bpe_cache_stats["known_pairs"]),
        "bpe_updated_pairs": int(bpe_cache_stats["updated_pairs"]),
        "bpe_shared_vocab_size": int(bpe_cache_stats["shared_vocab_size"]),
    }
    if manifest_items is not None:
        stats["manifest_count"] = len(manifest_items)
        stats["embedded_manifest"] = True

    if verbose:
        n0, n1, n2, n3, n4 = level_counts(original_size)
        suffix = (
            f", manifest={stats['manifest_count']}"
            if "manifest_count" in stats
            else f", gdict={db_path}"
        )
        print(f"Input:  {input_path} ({original_size} bytes)")
        print(
            f"Output: {output_path} ({out_size} bytes), "
            f"mode={stats['mode_name']}, ratio={stats['ratio']:.3f}x{suffix}, "
            f"snapshot={snapshot_meta['snapshot_id']}, "
            f"bpe-vocab={stats['bpe_shared_vocab_pairs_available']}, "
            f"applied={stats['bpe_shared_vocab_merges_applied']}"
        )
        print(f"Levels: L0={n0}, L1={n1}, L2={n2}, L3={n3}, L4={n4}")
    return stats


def _write_restored_from_level4_iter(
    output_path: Path,
    original_size: int,
    checksum: bytes,
    level4_iter,
) -> int:
    try:
        produced_bytes = 0
        produced_n4 = 0
        hasher = hashlib.blake2b(digest_size=16)
        out_f = output_path.open("wb")
        buf = array("I")
        for label in level4_iter:
            buf.append(int(label))
            produced_n4 += 1
            if len(buf) >= (1 << 15):
                produced_bytes = emit_decoded_l4_words(buf, out_f, hasher, original_size, produced_bytes)
                buf = array("I")
        if buf:
            produced_bytes = emit_decoded_l4_words(buf, out_f, hasher, original_size, produced_bytes)
        out_f.close()

        expected_n4 = level_counts(original_size)[4]
        if produced_n4 != expected_n4:
            raise ValueError(f"Corrupted container: expected {expected_n4} L4 labels, got {produced_n4}")
        if produced_bytes != original_size:
            raise ValueError(f"Corrupted container: expected {original_size} bytes, got {produced_bytes}")
        if hasher.digest() != checksum:
            raise ValueError("Checksum mismatch: decoded data is not identical to original")
        return produced_bytes
    except Exception:
        try:
            if 'out_f' in locals():
                out_f.close()
        except Exception:
            pass
        try:
            if output_path.exists():
                output_path.unlink()
        except Exception:
            pass
        raise


def decompress_file_bpe_gdict(
    input_path: Path,
    output_path: Path,
    mode_id: int,
    original_size: int,
    payload_size: int,
    checksum: bytes,
    payload_offset: int,
    global_dict_db: Path | None,
    container_metadata: dict | None = None,
    verbose: bool = True,
    workers: int = 1,
) -> dict:
    if mode_id != MODE_RANS_BPE_GLOBAL_DICT:
        raise ValueError("BPE global-dictionary decoder supports only ansbpegdict mode")
    if global_dict_db is None:
        raise ValueError("ansbpegdict mode requires --global-dict-db")

    db_path = Path(global_dict_db)
    conn = _open_global_dict_sqlite(db_path, readonly=True)
    try:
        container_global_dict_meta = _validate_global_dict_snapshot(
            conn,
            _extract_container_global_dict_meta(container_metadata),
        )
        payload = _read_container_payload_bytes(input_path, payload_offset, payload_size)
        gid_seq = decode_bpeans_sequence(payload, workers=workers)

        expected_n4 = level_counts(original_size)[4]
        if len(gid_seq) != expected_n4:
            raise ValueError(f"Corrupted container: expected {expected_n4} L4 labels, got {len(gid_seq)}")

        cache: dict[int, int] = {}
        cache_limit = 1 << 18
        level4: list[int] = []
        for gid in gid_seq:
            gid = int(gid)
            label = cache.get(gid)
            if label is None:
                row = conn.execute("SELECT label FROM global_dict WHERE gid=?", (gid,)).fetchone()
                if row is None:
                    raise ValueError(f"Global dictionary missing gid={gid}")
                label = int(row[0])
                if len(cache) >= cache_limit:
                    cache.clear()
                cache[gid] = label
            level4.append(label)
    finally:
        conn.close()

    data = restore_bytes_from_level4(level4, original_size)
    if len(data) != original_size:
        raise ValueError("Corrupted container: bad restored size")
    actual_checksum = hashlib.blake2b(data, digest_size=16).digest()
    if actual_checksum != checksum:
        raise ValueError("Checksum mismatch: decoded data is not identical to original")

    output_path.write_bytes(data)
    stats = {
        "output_size": len(data),
        "mode_id": mode_id,
        "mode_name": mode_name(mode_id),
        "streaming": False,
        "workers": max(1, int(workers)),
        "global_dict_db": str(db_path),
    }
    if container_global_dict_meta is not None:
        stats["global_dict_snapshot_id"] = int(container_global_dict_meta["snapshot_id"])
        stats["global_dict_db_uuid"] = str(container_global_dict_meta["db_uuid"])
    if verbose:
        suffix = (
            f", snapshot={stats['global_dict_snapshot_id']}"
            if "global_dict_snapshot_id" in stats
            else ""
        )
        print(f"Decompressed: {output_path} ({len(data)} bytes), mode={stats['mode_name']}, gdict={db_path}{suffix}")
    return stats


def decompress_file_bpe_gdictm(
    input_path: Path,
    output_path: Path,
    mode_id: int,
    original_size: int,
    payload_size: int,
    checksum: bytes,
    payload_offset: int,
    container_metadata: dict | None = None,
    verbose: bool = True,
    workers: int = 1,
) -> dict:
    if mode_id != MODE_RANS_BPE_GLOBAL_DICT_MANIFEST:
        raise ValueError("BPE global-manifest decoder supports only ansbpegdictm mode")

    payload = _read_container_payload_bytes(input_path, payload_offset, payload_size)
    container_global_dict_meta = _extract_container_global_dict_meta(container_metadata)
    manifest, gid_seq = _decode_global_bpe_manifest_payload(payload, workers=workers)

    if container_global_dict_meta is not None:
        embedded_flag = container_global_dict_meta.get("embedded_manifest", True)
        if not embedded_flag:
            raise ValueError("Container metadata mismatch: embedded manifest flag is false")
        meta_manifest_count = container_global_dict_meta.get("manifest_count")
        if meta_manifest_count is not None and int(meta_manifest_count) != len(manifest):
            raise ValueError("Container metadata mismatch: manifest_count differs from payload")

    expected_n4 = level_counts(original_size)[4]
    if len(gid_seq) != expected_n4:
        raise ValueError(f"Corrupted container: expected {expected_n4} L4 labels, got {len(gid_seq)}")

    try:
        level4 = [manifest[int(gid)] for gid in gid_seq]
    except KeyError as exc:
        raise ValueError("Corrupted BPE global-manifest payload: gid missing from manifest") from exc

    data = restore_bytes_from_level4(level4, original_size)
    if len(data) != original_size:
        raise ValueError("Corrupted container: bad restored size")
    actual_checksum = hashlib.blake2b(data, digest_size=16).digest()
    if actual_checksum != checksum:
        raise ValueError("Checksum mismatch: decoded data is not identical to original")

    output_path.write_bytes(data)
    stats = {
        "output_size": len(data),
        "mode_id": mode_id,
        "mode_name": mode_name(mode_id),
        "streaming": False,
        "workers": max(1, int(workers)),
        "embedded_manifest": True,
        "manifest_count": len(manifest),
    }
    if container_global_dict_meta is not None:
        stats["global_dict_snapshot_id"] = int(container_global_dict_meta["snapshot_id"])
        stats["global_dict_db_uuid"] = str(container_global_dict_meta["db_uuid"])
    if verbose:
        suffix = (
            f", snapshot={stats['global_dict_snapshot_id']}"
            if "global_dict_snapshot_id" in stats
            else ""
        )
        print(
            f"Decompressed: {output_path} ({len(data)} bytes), "
            f"mode={stats['mode_name']}, manifest={len(manifest)}{suffix}"
        )
    return stats


def compress_file_stream_bpe_gdict_sqlite(
    input_path: Path,
    output_path: Path,
    mode_id: int,
    chunk_size: int,
    global_dict_db: Path | None,
    workers: int = 1,
    metadata: dict | None = None,
    verbose: bool = True,
) -> dict:
    if mode_id not in (MODE_RANS_BPE_GLOBAL_DICT, MODE_RANS_BPE_GLOBAL_DICT_MANIFEST):
        raise ValueError("Streaming BPE global-dictionary encoder supports only ansbpegdict/ansbpegdictm modes")
    if global_dict_db is None:
        raise ValueError("ansbpegdict/ansbpegdictm modes require --global-dict-db for encoding")

    original_size = input_path.stat().st_size
    base_metadata = dict(metadata or {})
    expected_n4 = level_counts(original_size)[4]
    hasher = hashlib.blake2b(digest_size=16)
    db_path = Path(global_dict_db)

    with tempfile.TemporaryDirectory(prefix="vcomp_bpegdict_stream_") as td:
        td_path = Path(td)
        idx_path = td_path / "global_indices.bin"

        conn = _open_global_dict_sqlite(db_path, readonly=False)
        try:
            if mode_id == MODE_RANS_BPE_GLOBAL_DICT_MANIFEST:
                conn.execute("CREATE TEMP TABLE IF NOT EXISTS used_gid(gid INTEGER PRIMARY KEY)")
                conn.execute("DELETE FROM used_gid")
                conn.commit()
            conn.execute("BEGIN")
            l4_count = 0
            max_gid = 0
            cache: dict[int, int] = {}
            used_cache: set[int] = set()

            with input_path.open("rb") as in_f, idx_path.open("wb") as idx_f:
                for raw_chunk, words_cnt in _iter_l4_raw_chunks(
                    input_stream=in_f,
                    chunk_size=chunk_size,
                    workers=workers,
                    hasher=hasher,
                ):
                    arr = array("I")
                    arr.frombytes(raw_chunk)
                    if sys.byteorder != "little":
                        arr.byteswap()

                    l4_count += words_cnt
                    for label in arr:
                        gid = _assign_global_gid_sqlite(conn, cache, int(label))
                        if mode_id == MODE_RANS_BPE_GLOBAL_DICT_MANIFEST:
                            _mark_used_gid_sqlite(conn, used_cache, gid)
                        if gid > max_gid:
                            max_gid = gid
                        idx_f.write(struct.pack("<I", gid))

            conn.commit()

            if l4_count != expected_n4:
                raise ValueError(f"Internal error: L4 count mismatch {l4_count} != {expected_n4}")

            snapshot_meta = _create_global_dict_snapshot(conn, max_gid=max_gid)
            gid_seq = _read_u32_file_to_list(idx_path)
            if len(gid_seq) != expected_n4:
                raise ValueError(f"Internal error: gid count mismatch {len(gid_seq)} != {expected_n4}")
            seed_pairs = _load_global_bpe_seed_pairs(conn)
            bpe_cache_stats = _update_global_bpe_pairs(conn, gid_seq)
            conn.commit()

            container_metadata = dict(base_metadata)
            global_dict_meta = dict(snapshot_meta)
            manifest_items = None
            if mode_id == MODE_RANS_BPE_GLOBAL_DICT_MANIFEST:
                manifest_items = _collect_sqlite_global_manifest_items(conn)
                global_dict_meta["embedded_manifest"] = True
                global_dict_meta["manifest_count"] = len(manifest_items)
            container_metadata["global_dict"] = global_dict_meta
            container_metadata["bpe"] = {
                "persistent_cache": True,
                "persistent_vocabulary": True,
                "seed_pairs_available": len(seed_pairs),
                "shared_vocab_pairs_available": len(seed_pairs),
            }

            if mode_id == MODE_RANS_BPE_GLOBAL_DICT:
                payload, bpe_stats = encode_bpeans_sequence(gid_seq, seed_pairs=seed_pairs, workers=workers)
            else:
                inner_payload, bpe_stats = encode_bpeans_sequence(gid_seq, seed_pairs=seed_pairs, workers=workers)
                payload = _pack_global_bpe_manifest_payload(inner_payload, manifest_items or [])

            container_metadata["bpe"]["seed_merges_applied"] = int(bpe_stats["seed_merge_count"])
            container_metadata["bpe"]["shared_vocab_merges_applied"] = int(bpe_stats["shared_merge_count"])
            container_metadata["bpe"]["merge_count"] = int(bpe_stats["merge_count"])
            container_metadata["bpe"]["known_pairs"] = int(bpe_cache_stats["known_pairs"])
            container_metadata["bpe"]["updated_pairs"] = int(bpe_cache_stats["updated_pairs"])
            container_metadata["bpe"]["shared_vocab_size"] = int(bpe_cache_stats["shared_vocab_size"])

            checksum = hasher.digest()
            meta_blob = serialize_metadata(container_metadata)
            with output_path.open("wb") as out_f:
                out_f.write(
                    pack_header_v2(
                        mode_id=mode_id,
                        original_size=original_size,
                        payload_size=len(payload),
                        metadata_size=len(meta_blob),
                        checksum=checksum,
                    )
                )
                if meta_blob:
                    out_f.write(meta_blob)
                out_f.write(payload)

            out_size = output_path.stat().st_size
            stats = {
                "input_size": original_size,
                "output_size": out_size,
                "ratio": (original_size / out_size) if out_size else 0.0,
                "mode_id": mode_id,
                "mode_name": mode_name(mode_id),
                "n4": l4_count,
                "streaming": True,
                "workers": max(1, int(workers)),
                "global_dict_db": str(db_path),
                "global_dict_snapshot_id": int(snapshot_meta["snapshot_id"]),
                "global_dict_db_uuid": str(snapshot_meta["db_uuid"]),
                "global_dict_max_gid": int(snapshot_meta["max_gid"]),
                "global_dict_hash": str(snapshot_meta["dict_hash"]),
                "bpe_seed_pairs_available": len(seed_pairs),
                "bpe_seed_merges_applied": int(bpe_stats["seed_merge_count"]),
                "bpe_shared_vocab_pairs_available": len(seed_pairs),
                "bpe_shared_vocab_merges_applied": int(bpe_stats["shared_merge_count"]),
                "bpe_merge_count": int(bpe_stats["merge_count"]),
                "bpe_known_pairs": int(bpe_cache_stats["known_pairs"]),
                "bpe_updated_pairs": int(bpe_cache_stats["updated_pairs"]),
                "bpe_shared_vocab_size": int(bpe_cache_stats["shared_vocab_size"]),
            }
            if manifest_items is not None:
                stats["manifest_count"] = len(manifest_items)
                stats["embedded_manifest"] = True

            if verbose:
                n0, n1, n2, n3, n4 = level_counts(original_size)
                suffix = (
                    f", manifest={stats['manifest_count']}"
                    if "manifest_count" in stats
                    else f", gdict={db_path}"
                )
                print(f"Input:  {input_path} ({original_size} bytes)")
                print(
                    f"Output: {output_path} ({out_size} bytes), "
                    f"mode={stats['mode_name']}, ratio={stats['ratio']:.3f}x, stream=yes{suffix}, "
                    f"snapshot={snapshot_meta['snapshot_id']}, "
                    f"bpe-vocab={stats['bpe_shared_vocab_pairs_available']}, "
                    f"applied={stats['bpe_shared_vocab_merges_applied']}"
                )
                print(f"Levels: L0={n0}, L1={n1}, L2={n2}, L3={n3}, L4={n4}")
            return stats
        finally:
            conn.close()


def decompress_file_stream_bpe_gdict(
    input_path: Path,
    output_path: Path,
    mode_id: int,
    original_size: int,
    payload_size: int,
    checksum: bytes,
    payload_offset: int,
    global_dict_db: Path | None,
    container_metadata: dict | None = None,
    verbose: bool = True,
    workers: int = 1,
) -> dict:
    if mode_id != MODE_RANS_BPE_GLOBAL_DICT:
        raise ValueError("Streaming BPE global-dictionary decoder supports only ansbpegdict mode")
    if global_dict_db is None:
        raise ValueError("ansbpegdict mode requires --global-dict-db")

    db_path = Path(global_dict_db)
    conn = _open_global_dict_sqlite(db_path, readonly=True)
    try:
        container_global_dict_meta = _validate_global_dict_snapshot(
            conn,
            _extract_container_global_dict_meta(container_metadata),
        )
        gid_seq = decode_bpeans_sequence(
            _read_container_payload_bytes(input_path, payload_offset, payload_size),
            workers=workers,
        )

        cache: dict[int, int] = {}
        cache_limit = 1 << 18

        def label_iter():
            for gid in gid_seq:
                gid_i = int(gid)
                label = cache.get(gid_i)
                if label is None:
                    row = conn.execute("SELECT label FROM global_dict WHERE gid=?", (gid_i,)).fetchone()
                    if row is None:
                        raise ValueError(f"Global dictionary missing gid={gid_i}")
                    label = int(row[0])
                    if len(cache) >= cache_limit:
                        cache.clear()
                    cache[gid_i] = label
                yield label

        produced = _write_restored_from_level4_iter(output_path, original_size, checksum, label_iter())
    finally:
        conn.close()

    stats = {
        "output_size": produced,
        "mode_id": mode_id,
        "mode_name": mode_name(mode_id),
        "streaming": True,
        "workers": max(1, int(workers)),
        "global_dict_db": str(db_path),
    }
    if container_global_dict_meta is not None:
        stats["global_dict_snapshot_id"] = int(container_global_dict_meta["snapshot_id"])
        stats["global_dict_db_uuid"] = str(container_global_dict_meta["db_uuid"])
    if verbose:
        suffix = (
            f", snapshot={stats['global_dict_snapshot_id']}"
            if "global_dict_snapshot_id" in stats
            else ""
        )
        print(
            f"Decompressed: {output_path} ({produced} bytes), "
            f"mode={stats['mode_name']}, stream=yes, gdict={db_path}{suffix}"
        )
    return stats


def decompress_file_stream_bpe_gdictm(
    input_path: Path,
    output_path: Path,
    mode_id: int,
    original_size: int,
    payload_size: int,
    checksum: bytes,
    payload_offset: int,
    container_metadata: dict | None = None,
    verbose: bool = True,
    workers: int = 1,
) -> dict:
    if mode_id != MODE_RANS_BPE_GLOBAL_DICT_MANIFEST:
        raise ValueError("Streaming BPE global-manifest decoder supports only ansbpegdictm mode")

    container_global_dict_meta = _extract_container_global_dict_meta(container_metadata)
    manifest, gid_seq = _decode_global_bpe_manifest_payload(
        _read_container_payload_bytes(input_path, payload_offset, payload_size),
        workers=workers,
    )
    if container_global_dict_meta is not None:
        embedded_flag = container_global_dict_meta.get("embedded_manifest", True)
        if not embedded_flag:
            raise ValueError("Container metadata mismatch: embedded manifest flag is false")
        meta_manifest_count = container_global_dict_meta.get("manifest_count")
        if meta_manifest_count is not None and int(meta_manifest_count) != len(manifest):
            raise ValueError("Container metadata mismatch: manifest_count differs from payload")

    def label_iter():
        for gid in gid_seq:
            try:
                yield manifest[int(gid)]
            except KeyError as exc:
                raise ValueError("Corrupted BPE global-manifest payload: gid missing from manifest") from exc

    produced = _write_restored_from_level4_iter(output_path, original_size, checksum, label_iter())
    stats = {
        "output_size": produced,
        "mode_id": mode_id,
        "mode_name": mode_name(mode_id),
        "streaming": True,
        "workers": max(1, int(workers)),
        "embedded_manifest": True,
        "manifest_count": len(manifest),
    }
    if container_global_dict_meta is not None:
        stats["global_dict_snapshot_id"] = int(container_global_dict_meta["snapshot_id"])
        stats["global_dict_db_uuid"] = str(container_global_dict_meta["db_uuid"])
    if verbose:
        suffix = (
            f", snapshot={stats['global_dict_snapshot_id']}"
            if "global_dict_snapshot_id" in stats
            else ""
        )
        print(
            f"Decompressed: {output_path} ({produced} bytes), "
            f"mode={stats['mode_name']}, stream=yes, manifest={len(manifest)}{suffix}"
        )
    return stats


def compress_file_stream_dict_or_zdict_sqlite(
    input_path: Path,
    output_path: Path,
    mode_id: int,
    chunk_size: int,
    workers: int = 1,
    metadata: dict | None = None,
    verbose: bool = True,
) -> dict:
    if mode_id not in (MODE_DICT_BITPACK, MODE_ZLIB_DICT_BITPACK, MODE_RANS_DICT_BITPACK):
        raise ValueError("Streaming dict encoder supports only dict/zdict/ansdict modes")

    original_size = input_path.stat().st_size
    meta_blob = serialize_metadata(metadata)
    expected_n4 = level_counts(original_size)[4]
    hasher = hashlib.blake2b(digest_size=16)

    with tempfile.TemporaryDirectory(prefix="vcomp_dict_stream_") as td:
        td_path = Path(td)
        db_path = td_path / "dict.sqlite"
        idx_path = td_path / "indices.bin"

        conn = sqlite3.connect(db_path)
        try:
            conn.execute("PRAGMA journal_mode=OFF")
            conn.execute("PRAGMA synchronous=OFF")
            conn.execute("PRAGMA temp_store=MEMORY")
            conn.execute("PRAGMA locking_mode=EXCLUSIVE")
            conn.execute("CREATE TABLE dict_map(label INTEGER PRIMARY KEY, idx INTEGER NOT NULL UNIQUE)")
            conn.execute("BEGIN")

            next_idx = 0
            l4_count = 0
            cache: dict[int, int] = {}

            with input_path.open("rb") as in_f, idx_path.open("wb") as idx_f:
                for raw_chunk, words_cnt in _iter_l4_raw_chunks(
                    input_stream=in_f,
                    chunk_size=chunk_size,
                    workers=workers,
                    hasher=hasher,
                ):
                    arr = array("I")
                    arr.frombytes(raw_chunk)
                    if sys.byteorder != "little":
                        arr.byteswap()

                    l4_count += words_cnt
                    for label in arr:
                        idx, next_idx = _assign_idx_sqlite(conn, cache, int(label), next_idx)
                        idx_f.write(struct.pack("<I", idx))

            conn.commit()

            if l4_count != expected_n4:
                raise ValueError(f"Internal error: L4 count mismatch {l4_count} != {expected_n4}")

            dict_count = next_idx
            index_count = l4_count
            bit_width = bits_required(dict_count)
            packed_size = (index_count * bit_width + 7) // 8
            payload_head = struct.pack("<IBII", dict_count, bit_width, index_count, packed_size)

            if mode_id == MODE_RANS_DICT_BITPACK:
                with output_path.open("wb") as out_f:
                    out_f.write(b"\x00" * (HEADER_V2_SIZE + len(meta_blob)))
                    encoder = FramedRansEncoder(out_f.write, block_size=max(1 << 16, chunk_size))
                    encoder.feed(payload_head)
                    _write_sqlite_dictionary_blob(conn, encoder.feed)
                    with idx_path.open("rb") as idx_f:
                        wrote_packed = bitpack_u32_file_to_writer(
                            idx_f,
                            index_count=index_count,
                            bit_width=bit_width,
                            write_bytes=encoder.feed,
                        )
                    if wrote_packed != packed_size:
                        raise ValueError(
                            f"Internal error: packed size mismatch {wrote_packed} != {packed_size}"
                        )
                    payload_size = encoder.finish()

                    checksum = hasher.digest()
                    header = pack_header_v2(
                        mode_id=mode_id,
                        original_size=original_size,
                        payload_size=payload_size,
                        metadata_size=len(meta_blob),
                        checksum=checksum,
                    )
                    out_f.seek(0)
                    out_f.write(header)
                    if meta_blob:
                        out_f.write(meta_blob)
            else:
                payload_size = 0
                compressor = zlib.compressobj(level=9) if mode_id == MODE_ZLIB_DICT_BITPACK else None

                with output_path.open("wb") as out_f:
                    out_f.write(b"\x00" * (HEADER_V2_SIZE + len(meta_blob)))

                    def emit(raw_bytes: bytes) -> None:
                        nonlocal payload_size
                        if compressor is None:
                            out_f.write(raw_bytes)
                            payload_size += len(raw_bytes)
                        else:
                            comp = compressor.compress(raw_bytes)
                            if comp:
                                out_f.write(comp)
                                payload_size += len(comp)

                    emit(payload_head)
                    _write_sqlite_dictionary_blob(conn, emit)

                    with idx_path.open("rb") as idx_f:
                        wrote_packed = bitpack_u32_file_to_writer(
                            idx_f,
                            index_count=index_count,
                            bit_width=bit_width,
                            write_bytes=emit,
                        )
                    if wrote_packed != packed_size:
                        raise ValueError(
                            f"Internal error: packed size mismatch {wrote_packed} != {packed_size}"
                        )

                    if compressor is not None:
                        tail = compressor.flush()
                        if tail:
                            out_f.write(tail)
                            payload_size += len(tail)

                    checksum = hasher.digest()
                    header = pack_header_v2(
                        mode_id=mode_id,
                        original_size=original_size,
                        payload_size=payload_size,
                        metadata_size=len(meta_blob),
                        checksum=checksum,
                    )
                    out_f.seek(0)
                    out_f.write(header)
                    if meta_blob:
                        out_f.write(meta_blob)

            out_size = output_path.stat().st_size
            stats = {
                "input_size": original_size,
                "output_size": out_size,
                "ratio": (original_size / out_size) if out_size else 0.0,
                "mode_id": mode_id,
                "mode_name": mode_name(mode_id),
                "n4": l4_count,
                "streaming": True,
                "dict_backend": "sqlite",
            }

            if verbose:
                n0, n1, n2, n3, n4 = level_counts(original_size)
                print(f"Input:  {input_path} ({original_size} bytes)")
                print(
                    f"Output: {output_path} ({out_size} bytes), "
                    f"mode={stats['mode_name']}, ratio={stats['ratio']:.3f}x, stream=yes, dict=sqlite"
                )
                print(f"Levels: L0={n0}, L1={n1}, L2={n2}, L3={n3}, L4={n4}")
            return stats
        finally:
            conn.close()


def compress_file_stream_gdict_or_zgdict_sqlite(
    input_path: Path,
    output_path: Path,
    mode_id: int,
    chunk_size: int,
    global_dict_db: Path | None,
    workers: int = 1,
    metadata: dict | None = None,
    verbose: bool = True,
) -> dict:
    if mode_id not in (MODE_GLOBAL_DICT, MODE_ZGLOBAL_DICT, MODE_RANS_GLOBAL_DICT):
        raise ValueError("Streaming global dict encoder supports only gdict/zgdict/ansgdict modes")
    if global_dict_db is None:
        raise ValueError("gdict/zgdict modes require --global-dict-db")

    original_size = input_path.stat().st_size
    base_metadata = dict(metadata or {})
    expected_n4 = level_counts(original_size)[4]
    hasher = hashlib.blake2b(digest_size=16)
    db_path = Path(global_dict_db)

    with tempfile.TemporaryDirectory(prefix="vcomp_gdict_stream_") as td:
        td_path = Path(td)
        idx_path = td_path / "global_indices.bin"

        conn = _open_global_dict_sqlite(db_path, readonly=False)
        try:
            conn.execute("BEGIN")
            l4_count = 0
            max_gid = 0
            cache: dict[int, int] = {}

            with input_path.open("rb") as in_f, idx_path.open("wb") as idx_f:
                for raw_chunk, words_cnt in _iter_l4_raw_chunks(
                    input_stream=in_f,
                    chunk_size=chunk_size,
                    workers=workers,
                    hasher=hasher,
                ):
                    arr = array("I")
                    arr.frombytes(raw_chunk)
                    if sys.byteorder != "little":
                        arr.byteswap()

                    l4_count += words_cnt
                    for label in arr:
                        gid = _assign_global_gid_sqlite(conn, cache, int(label))
                        if gid > max_gid:
                            max_gid = gid
                        idx_f.write(struct.pack("<I", gid))

            conn.commit()

            if l4_count != expected_n4:
                raise ValueError(f"Internal error: L4 count mismatch {l4_count} != {expected_n4}")

            snapshot_meta = _create_global_dict_snapshot(conn, max_gid=max_gid)
            container_metadata = dict(base_metadata)
            container_metadata["global_dict"] = snapshot_meta
            meta_blob = serialize_metadata(container_metadata)

            index_count = l4_count
            bit_width = max(1, int(max_gid).bit_length())
            packed_size = (index_count * bit_width + 7) // 8
            payload_head = struct.pack("<BII", bit_width, index_count, packed_size)

            if mode_id == MODE_GLOBAL_DICT:
                entropy_mode = "raw"
            elif mode_id == MODE_ZGLOBAL_DICT:
                entropy_mode = "zlib"
            else:
                entropy_mode = "rans"

            with output_path.open("wb") as out_f:
                out_f.write(b"\x00" * (HEADER_V2_SIZE + len(meta_blob)))
                if entropy_mode == "rans":
                    encoder = FramedRansEncoder(out_f.write, block_size=max(1 << 16, chunk_size))
                    encoder.feed(payload_head)
                    with idx_path.open("rb") as idx_f:
                        wrote_packed = bitpack_u32_file_to_writer(
                            idx_f,
                            index_count=index_count,
                            bit_width=bit_width,
                            write_bytes=encoder.feed,
                        )
                    if wrote_packed != packed_size:
                        raise ValueError(
                            f"Internal error: packed size mismatch {wrote_packed} != {packed_size}"
                        )
                    payload_size = encoder.finish()
                else:
                    raw_payload_path = td_path / "gdict_payload.bin"
                    with raw_payload_path.open("wb") as raw_f:
                        raw_f.write(payload_head)
                        with idx_path.open("rb") as idx_f:
                            wrote_packed = bitpack_u32_file_to_writer(
                                idx_f,
                                index_count=index_count,
                                bit_width=bit_width,
                                write_bytes=raw_f.write,
                            )
                        if wrote_packed != packed_size:
                            raise ValueError(
                                f"Internal error: packed size mismatch {wrote_packed} != {packed_size}"
                            )
                    payload_size = _emit_temp_payload_file(raw_payload_path, out_f, entropy_mode=entropy_mode)

                checksum = hasher.digest()
                header = pack_header_v2(
                    mode_id=mode_id,
                    original_size=original_size,
                    payload_size=payload_size,
                    metadata_size=len(meta_blob),
                    checksum=checksum,
                )
                out_f.seek(0)
                out_f.write(header)
                if meta_blob:
                    out_f.write(meta_blob)

            out_size = output_path.stat().st_size
            stats = {
                "input_size": original_size,
                "output_size": out_size,
                "ratio": (original_size / out_size) if out_size else 0.0,
                "mode_id": mode_id,
                "mode_name": mode_name(mode_id),
                "n4": l4_count,
                "streaming": True,
                "global_dict_db": str(db_path),
                "global_dict_snapshot_id": int(snapshot_meta["snapshot_id"]),
                "global_dict_db_uuid": str(snapshot_meta["db_uuid"]),
                "global_dict_max_gid": int(snapshot_meta["max_gid"]),
                "global_dict_hash": str(snapshot_meta["dict_hash"]),
            }
            if verbose:
                n0, n1, n2, n3, n4 = level_counts(original_size)
                print(f"Input:  {input_path} ({original_size} bytes)")
                print(
                    f"Output: {output_path} ({out_size} bytes), "
                    f"mode={stats['mode_name']}, ratio={stats['ratio']:.3f}x, "
                    f"stream=yes, gdict={db_path}, snapshot={snapshot_meta['snapshot_id']}"
                )
                print(f"Levels: L0={n0}, L1={n1}, L2={n2}, L3={n3}, L4={n4}")
            return stats
        finally:
            conn.close()


def compress_file_stream_gdictm_or_zgdictm_sqlite(
    input_path: Path,
    output_path: Path,
    mode_id: int,
    chunk_size: int,
    global_dict_db: Path | None,
    workers: int = 1,
    metadata: dict | None = None,
    verbose: bool = True,
) -> dict:
    if mode_id not in (
        MODE_GLOBAL_DICT_MANIFEST,
        MODE_ZGLOBAL_DICT_MANIFEST,
        MODE_RANS_GLOBAL_DICT_MANIFEST,
    ):
        raise ValueError(
            "Streaming global-manifest encoder supports only gdictm/zgdictm/ansgdictm modes"
        )
    if global_dict_db is None:
        raise ValueError("gdictm/zgdictm modes require --global-dict-db for encoding")

    original_size = input_path.stat().st_size
    base_metadata = dict(metadata or {})
    expected_n4 = level_counts(original_size)[4]
    hasher = hashlib.blake2b(digest_size=16)
    db_path = Path(global_dict_db)

    with tempfile.TemporaryDirectory(prefix="vcomp_gdictm_stream_") as td:
        td_path = Path(td)
        idx_path = td_path / "global_indices.bin"

        conn = _open_global_dict_sqlite(db_path, readonly=False)
        try:
            conn.execute("CREATE TEMP TABLE IF NOT EXISTS used_gid(gid INTEGER PRIMARY KEY)")
            conn.execute("DELETE FROM used_gid")
            conn.commit()
            conn.execute("BEGIN")
            l4_count = 0
            max_gid = 0
            gid_cache: dict[int, int] = {}
            used_cache: set[int] = set()

            with input_path.open("rb") as in_f, idx_path.open("wb") as idx_f:
                for raw_chunk, words_cnt in _iter_l4_raw_chunks(
                    input_stream=in_f,
                    chunk_size=chunk_size,
                    workers=workers,
                    hasher=hasher,
                ):
                    arr = array("I")
                    arr.frombytes(raw_chunk)
                    if sys.byteorder != "little":
                        arr.byteswap()

                    l4_count += words_cnt
                    for label in arr:
                        gid = _assign_global_gid_sqlite(conn, gid_cache, int(label))
                        _mark_used_gid_sqlite(conn, used_cache, gid)
                        if gid > max_gid:
                            max_gid = gid
                        idx_f.write(struct.pack("<I", gid))

            conn.commit()

            if l4_count != expected_n4:
                raise ValueError(f"Internal error: L4 count mismatch {l4_count} != {expected_n4}")

            snapshot_meta = _create_global_dict_snapshot(conn, max_gid=max_gid)
            manifest_count = int(conn.execute("SELECT COUNT(*) FROM used_gid").fetchone()[0])
            index_count = l4_count
            bit_width = max(1, int(max_gid).bit_length())
            packed_size = (index_count * bit_width + 7) // 8

            container_metadata = dict(base_metadata)
            global_dict_meta = dict(snapshot_meta)
            global_dict_meta["embedded_manifest"] = True
            global_dict_meta["manifest_count"] = manifest_count
            container_metadata["global_dict"] = global_dict_meta
            meta_blob = serialize_metadata(container_metadata)

            payload_head = struct.pack("<III", manifest_count, index_count, packed_size)

            if mode_id == MODE_GLOBAL_DICT_MANIFEST:
                entropy_mode = "raw"
            elif mode_id == MODE_ZGLOBAL_DICT_MANIFEST:
                entropy_mode = "zlib"
            else:
                entropy_mode = "rans"

            with output_path.open("wb") as out_f:
                out_f.write(b"\x00" * (HEADER_V2_SIZE + len(meta_blob)))
                if entropy_mode == "rans":
                    encoder = FramedRansEncoder(out_f.write, block_size=max(1 << 16, chunk_size))
                    encoder.feed(payload_head)
                    manifest_bytes, wrote_manifest_count = _write_sqlite_global_manifest_blob(
                        conn,
                        encoder.feed,
                    )
                    if wrote_manifest_count != manifest_count or manifest_bytes != manifest_count * 8:
                        raise ValueError("Internal error: embedded manifest size mismatch")

                    with idx_path.open("rb") as idx_f:
                        wrote_packed = bitpack_u32_file_to_writer(
                            idx_f,
                            index_count=index_count,
                            bit_width=bit_width,
                            write_bytes=encoder.feed,
                        )
                    if wrote_packed != packed_size:
                        raise ValueError(
                            f"Internal error: packed size mismatch {wrote_packed} != {packed_size}"
                        )
                    payload_size = encoder.finish()
                else:
                    raw_payload_path = td_path / "gdictm_payload.bin"
                    with raw_payload_path.open("wb") as raw_f:
                        raw_f.write(payload_head)
                        manifest_bytes, wrote_manifest_count = _write_sqlite_global_manifest_blob(conn, raw_f.write)
                        if wrote_manifest_count != manifest_count or manifest_bytes != manifest_count * 8:
                            raise ValueError("Internal error: embedded manifest size mismatch")

                        with idx_path.open("rb") as idx_f:
                            wrote_packed = bitpack_u32_file_to_writer(
                                idx_f,
                                index_count=index_count,
                                bit_width=bit_width,
                                write_bytes=raw_f.write,
                            )
                        if wrote_packed != packed_size:
                            raise ValueError(
                                f"Internal error: packed size mismatch {wrote_packed} != {packed_size}"
                            )
                    payload_size = _emit_temp_payload_file(raw_payload_path, out_f, entropy_mode=entropy_mode)

                checksum = hasher.digest()
                header = pack_header_v2(
                    mode_id=mode_id,
                    original_size=original_size,
                    payload_size=payload_size,
                    metadata_size=len(meta_blob),
                    checksum=checksum,
                )
                out_f.seek(0)
                out_f.write(header)
                if meta_blob:
                    out_f.write(meta_blob)

            out_size = output_path.stat().st_size
            stats = {
                "input_size": original_size,
                "output_size": out_size,
                "ratio": (original_size / out_size) if out_size else 0.0,
                "mode_id": mode_id,
                "mode_name": mode_name(mode_id),
                "n4": l4_count,
                "streaming": True,
                "global_dict_db": str(db_path),
                "global_dict_snapshot_id": int(snapshot_meta["snapshot_id"]),
                "global_dict_db_uuid": str(snapshot_meta["db_uuid"]),
                "global_dict_max_gid": int(snapshot_meta["max_gid"]),
                "global_dict_hash": str(snapshot_meta["dict_hash"]),
                "manifest_count": manifest_count,
                "embedded_manifest": True,
            }
            if verbose:
                n0, n1, n2, n3, n4 = level_counts(original_size)
                print(f"Input:  {input_path} ({original_size} bytes)")
                print(
                    f"Output: {output_path} ({out_size} bytes), "
                    f"mode={stats['mode_name']}, ratio={stats['ratio']:.3f}x, "
                    f"stream=yes, manifest={manifest_count}, snapshot={snapshot_meta['snapshot_id']}"
                )
                print(f"Levels: L0={n0}, L1={n1}, L2={n2}, L3={n3}, L4={n4}")
            return stats
        finally:
            conn.close()


class DictPayloadStreamingDecoder:
    def __init__(self, output_path: Path, original_size: int, expected_checksum: bytes):
        self.output_path = output_path
        self.original_size = original_size
        self.expected_checksum = expected_checksum
        self.expected_n4 = level_counts(original_size)[4]

        self._out_f = output_path.open("wb")
        self._hasher = hashlib.blake2b(digest_size=16)
        self._out_buf = bytearray()

        self._head_size = struct.calcsize("<IBII")
        self._head_buf = bytearray()
        self._header_parsed = False

        self.dict_count = 0
        self.bit_width = 0
        self.index_count = 0
        self.packed_size = 0
        self.mask = 0

        self._dict_bytes_remaining = 0
        self._dict_carry = b""
        self._dictionary = array("I")

        self._packed_bytes_remaining = 0
        self._bitbuf = 0
        self._bits = 0

        self._produced_idx = 0
        self._produced_bytes = 0

        self._total_in = 0
        self._expected_payload_bytes: int | None = None
        self._closed = False

    def _flush_out(self) -> None:
        if self._out_buf:
            self._out_f.write(self._out_buf)
            self._hasher.update(self._out_buf)
            self._out_buf.clear()

    def _emit_label(self, label: int) -> None:
        remaining = self.original_size - self._produced_bytes
        if remaining <= 0:
            return

        b1 = (label >> 24) & 0xFF
        b2 = (label >> 16) & 0xFF
        b3 = (label >> 8) & 0xFF
        b4 = label & 0xFF

        if remaining >= 4:
            self._out_buf.extend((b1, b2, b3, b4))
            self._produced_bytes += 4
        else:
            if remaining == 3:
                self._out_buf.extend((b1, b2, b3))
            elif remaining == 2:
                self._out_buf.extend((b1, b2))
            else:
                self._out_buf.append(b1)
            self._produced_bytes += remaining

        if len(self._out_buf) >= (1 << 20):
            self._flush_out()

    def feed(self, data: bytes | memoryview) -> None:
        if self._closed:
            raise ValueError("Decoder already closed")
        if not data:
            return

        mv = memoryview(data)
        self._total_in += len(mv)
        pos = 0
        mv_len = len(mv)

        while pos < mv_len:
            if not self._header_parsed:
                need = self._head_size - len(self._head_buf)
                take = min(need, mv_len - pos)
                self._head_buf.extend(mv[pos : pos + take])
                pos += take
                if len(self._head_buf) == self._head_size:
                    self.dict_count, self.bit_width, self.index_count, self.packed_size = struct.unpack(
                        "<IBII", self._head_buf
                    )

                    if not (1 <= self.bit_width <= 32):
                        raise ValueError("Invalid dict-bitpack payload (bad bit width)")
                    if self.dict_count == 0 and self.index_count > 0:
                        raise ValueError("Invalid dict-bitpack payload (empty dict with non-empty index stream)")
                    if self.index_count != self.expected_n4:
                        raise ValueError(
                            f"Corrupted container: expected {self.expected_n4} L4 labels, got {self.index_count}"
                        )

                    self.mask = (1 << self.bit_width) - 1
                    self._dict_bytes_remaining = self.dict_count * 4
                    self._packed_bytes_remaining = self.packed_size
                    self._expected_payload_bytes = self._head_size + self._dict_bytes_remaining + self._packed_bytes_remaining
                    self._header_parsed = True
                continue

            if self._dict_bytes_remaining > 0:
                take = min(self._dict_bytes_remaining, mv_len - pos)
                part = bytes(mv[pos : pos + take])
                pos += take
                self._dict_bytes_remaining -= take

                tmp = self._dict_carry + part
                complete = (len(tmp) // 4) * 4
                if complete:
                    arr = array("I")
                    arr.frombytes(tmp[:complete])
                    if sys.byteorder != "little":
                        arr.byteswap()
                    self._dictionary.extend(arr)
                self._dict_carry = tmp[complete:]

                if self._dict_bytes_remaining == 0 and self._dict_carry:
                    raise ValueError("Invalid dict payload alignment")
                continue

            if self._packed_bytes_remaining > 0:
                take = min(self._packed_bytes_remaining, mv_len - pos)
                part = mv[pos : pos + take]
                pos += take
                self._packed_bytes_remaining -= take

                for byte in part:
                    self._bitbuf |= int(byte) << self._bits
                    self._bits += 8
                    while self._bits >= self.bit_width and self._produced_idx < self.index_count:
                        idx = self._bitbuf & self.mask
                        self._bitbuf >>= self.bit_width
                        self._bits -= self.bit_width

                        if idx >= self.dict_count:
                            raise ValueError("Corrupted index stream: dictionary index out of range")
                        label = int(self._dictionary[idx])
                        self._produced_idx += 1
                        self._emit_label(label)
                continue

            # Header+dict+packed already consumed, but more bytes appeared.
            raise ValueError("Invalid dict-bitpack payload (extra bytes)")

    def finish(self) -> int:
        if self._closed:
            raise ValueError("Decoder already closed")

        try:
            if not self._header_parsed:
                raise ValueError("Invalid dict-bitpack payload (missing header)")
            if self._dict_bytes_remaining != 0:
                raise ValueError("Invalid dict-bitpack payload (dictionary truncated)")
            if self._packed_bytes_remaining != 0:
                raise ValueError("Invalid dict-bitpack payload (packed stream truncated)")
            if self._produced_idx != self.index_count:
                raise ValueError(
                    f"Corrupted container: expected {self.index_count} indices, got {self._produced_idx}"
                )
            if self._produced_bytes != self.original_size:
                raise ValueError(
                    f"Corrupted container: expected {self.original_size} bytes, got {self._produced_bytes}"
                )
            if self._expected_payload_bytes is not None and self._total_in != self._expected_payload_bytes:
                raise ValueError("Invalid dict-bitpack payload (size mismatch)")

            self._flush_out()
            self._out_f.close()
            self._closed = True

            if self._hasher.digest() != self.expected_checksum:
                raise ValueError("Checksum mismatch: decoded data is not identical to original")
            return self._produced_bytes
        except Exception:
            self.abort()
            raise

    def abort(self) -> None:
        if not self._closed:
            try:
                self._out_f.close()
            finally:
                self._closed = True
            try:
                if self.output_path.exists():
                    self.output_path.unlink()
            except Exception:
                pass


def decompress_file_stream_dict_or_zdict(
    input_path: Path,
    output_path: Path,
    mode_id: int,
    original_size: int,
    payload_size: int,
    checksum: bytes,
    payload_offset: int,
    chunk_size: int,
    verbose: bool = True,
) -> dict:
    if mode_id not in (MODE_DICT_BITPACK, MODE_ZLIB_DICT_BITPACK, MODE_RANS_DICT_BITPACK):
        raise ValueError("Streaming dict decoder supports only dict/zdict/ansdict modes")

    decoder = DictPayloadStreamingDecoder(
        output_path=output_path,
        original_size=original_size,
        expected_checksum=checksum,
    )

    try:
        if mode_id == MODE_DICT_BITPACK:
            with input_path.open("rb") as in_f:
                # mmap path for large uncompressed dict payloads
                if payload_size >= (64 * 1024 * 1024):
                    mm = mmap.mmap(in_f.fileno(), 0, access=mmap.ACCESS_READ)
                    try:
                        start = payload_offset
                        end = payload_offset + payload_size
                        pos = start
                        while pos < end:
                            nxt = min(pos + chunk_size, end)
                            decoder.feed(memoryview(mm)[pos:nxt])
                            pos = nxt
                    finally:
                        mm.close()
                else:
                    in_f.seek(payload_offset)
                    remaining = payload_size
                    while remaining > 0:
                        chunk = in_f.read(min(chunk_size, remaining))
                        if not chunk:
                            raise ValueError("Invalid container: payload truncated")
                        remaining -= len(chunk)
                        decoder.feed(chunk)
        elif mode_id == MODE_ZLIB_DICT_BITPACK:
            with input_path.open("rb") as in_f:
                in_f.seek(payload_offset)
                decomp = zlib.decompressobj()
                remaining = payload_size
                while remaining > 0:
                    comp_chunk = in_f.read(min(chunk_size, remaining))
                    if not comp_chunk:
                        raise ValueError("Invalid container: payload truncated")
                    remaining -= len(comp_chunk)

                    dec = decomp.decompress(comp_chunk)
                    if dec:
                        decoder.feed(dec)

                tail = decomp.flush()
                if tail:
                    decoder.feed(tail)
                if decomp.unused_data:
                    raise ValueError("Corrupted zdict payload: trailing compressed data")
        else:
            rans_decoder = AutoRansStreamingDecoder(decoder.feed)
            with input_path.open("rb") as in_f:
                in_f.seek(payload_offset)
                remaining = payload_size
                while remaining > 0:
                    comp_chunk = in_f.read(min(chunk_size, remaining))
                    if not comp_chunk:
                        raise ValueError("Invalid container: payload truncated")
                    remaining -= len(comp_chunk)
                    rans_decoder.feed(comp_chunk)
            rans_decoder.finish()

        produced = decoder.finish()
    except Exception:
        decoder.abort()
        raise

    stats = {
        "output_size": produced,
        "mode_id": mode_id,
        "mode_name": mode_name(mode_id),
        "streaming": True,
    }
    if verbose:
        print(f"Decompressed: {output_path} ({produced} bytes), mode={stats['mode_name']}, stream=yes")
    return stats


class GlobalDictPayloadStreamingDecoder:
    def __init__(
        self,
        output_path: Path,
        original_size: int,
        expected_checksum: bytes,
        global_dict_db: Path,
        container_global_dict_meta: dict | None = None,
    ):
        self.output_path = output_path
        self.original_size = original_size
        self.expected_checksum = expected_checksum
        self.expected_n4 = level_counts(original_size)[4]

        self._head_size = struct.calcsize("<BII")
        self._head_buf = bytearray()
        self._header_parsed = False

        self.bit_width = 0
        self.index_count = 0
        self.packed_size = 0
        self.mask = 0

        self._packed_bytes_remaining = 0
        self._bitbuf = 0
        self._bits = 0

        self._produced_idx = 0
        self._produced_bytes = 0

        self._total_in = 0
        self._expected_payload_bytes: int | None = None
        self._closed = False

        self._conn = None
        self._out_f = None
        self._hasher = None
        self._out_buf = bytearray()
        try:
            self._conn = _open_global_dict_sqlite(Path(global_dict_db), readonly=True)
            self._container_global_dict_meta = _validate_global_dict_snapshot(
                self._conn,
                container_global_dict_meta,
            )
            self._out_f = output_path.open("wb")
            self._hasher = hashlib.blake2b(digest_size=16)
        except Exception:
            if self._out_f is not None:
                try:
                    self._out_f.close()
                except Exception:
                    pass
            if self._conn is not None:
                try:
                    self._conn.close()
                except Exception:
                    pass
            raise
        self._cache: dict[int, int] = {}
        self._cache_limit = 1 << 18

    def _flush_out(self) -> None:
        if self._out_buf:
            self._out_f.write(self._out_buf)
            self._hasher.update(self._out_buf)
            self._out_buf.clear()

    def _emit_label(self, label: int) -> None:
        remaining = self.original_size - self._produced_bytes
        if remaining <= 0:
            return

        b1 = (label >> 24) & 0xFF
        b2 = (label >> 16) & 0xFF
        b3 = (label >> 8) & 0xFF
        b4 = label & 0xFF

        if remaining >= 4:
            self._out_buf.extend((b1, b2, b3, b4))
            self._produced_bytes += 4
        else:
            if remaining == 3:
                self._out_buf.extend((b1, b2, b3))
            elif remaining == 2:
                self._out_buf.extend((b1, b2))
            else:
                self._out_buf.append(b1)
            self._produced_bytes += remaining

        if len(self._out_buf) >= (1 << 20):
            self._flush_out()

    def _resolve_label(self, gid: int) -> int:
        label = self._cache.get(gid)
        if label is not None:
            return label

        row = self._conn.execute("SELECT label FROM global_dict WHERE gid=?", (gid,)).fetchone()
        if row is None:
            raise ValueError(f"Global dictionary missing gid={gid}")
        label = int(row[0])

        if len(self._cache) >= self._cache_limit:
            self._cache.clear()
        self._cache[gid] = label
        return label

    def feed(self, data: bytes | memoryview) -> None:
        if self._closed:
            raise ValueError("Decoder already closed")
        if not data:
            return

        mv = memoryview(data)
        self._total_in += len(mv)
        pos = 0
        mv_len = len(mv)

        while pos < mv_len:
            if not self._header_parsed:
                need = self._head_size - len(self._head_buf)
                take = min(need, mv_len - pos)
                self._head_buf.extend(mv[pos : pos + take])
                pos += take
                if len(self._head_buf) == self._head_size:
                    self.bit_width, self.index_count, self.packed_size = struct.unpack(
                        "<BII", self._head_buf
                    )
                    if not (1 <= self.bit_width <= 32):
                        raise ValueError("Invalid global-dict payload (bad bit width)")
                    if self.index_count != self.expected_n4:
                        raise ValueError(
                            f"Corrupted container: expected {self.expected_n4} L4 labels, got {self.index_count}"
                        )
                    self.mask = (1 << self.bit_width) - 1
                    self._packed_bytes_remaining = self.packed_size
                    self._expected_payload_bytes = self._head_size + self._packed_bytes_remaining
                    self._header_parsed = True
                continue

            if self._packed_bytes_remaining > 0:
                take = min(self._packed_bytes_remaining, mv_len - pos)
                part = mv[pos : pos + take]
                pos += take
                self._packed_bytes_remaining -= take

                for byte in part:
                    self._bitbuf |= int(byte) << self._bits
                    self._bits += 8
                    while self._bits >= self.bit_width and self._produced_idx < self.index_count:
                        gid = self._bitbuf & self.mask
                        self._bitbuf >>= self.bit_width
                        self._bits -= self.bit_width
                        label = self._resolve_label(int(gid))
                        self._produced_idx += 1
                        self._emit_label(label)
                continue

            raise ValueError("Invalid global-dict payload (extra bytes)")

    def finish(self) -> int:
        if self._closed:
            raise ValueError("Decoder already closed")

        try:
            if not self._header_parsed:
                raise ValueError("Invalid global-dict payload (missing header)")
            if self._packed_bytes_remaining != 0:
                raise ValueError("Invalid global-dict payload (packed stream truncated)")
            if self._produced_idx != self.index_count:
                raise ValueError(
                    f"Corrupted container: expected {self.index_count} indices, got {self._produced_idx}"
                )
            if self._produced_bytes != self.original_size:
                raise ValueError(
                    f"Corrupted container: expected {self.original_size} bytes, got {self._produced_bytes}"
                )
            if self._expected_payload_bytes is not None and self._total_in != self._expected_payload_bytes:
                raise ValueError("Invalid global-dict payload (size mismatch)")

            self._flush_out()
            self._out_f.close()
            self._conn.close()
            self._closed = True

            if self._hasher.digest() != self.expected_checksum:
                raise ValueError("Checksum mismatch: decoded data is not identical to original")
            return self._produced_bytes
        except Exception:
            self.abort()
            raise

    def abort(self) -> None:
        if not self._closed:
            try:
                if self._out_f is not None:
                    self._out_f.close()
            finally:
                self._closed = True
            try:
                if self._conn is not None:
                    self._conn.close()
            except Exception:
                pass
            try:
                if self.output_path.exists():
                    self.output_path.unlink()
            except Exception:
                pass


class GlobalDictManifestPayloadStreamingDecoder:
    def __init__(
        self,
        output_path: Path,
        original_size: int,
        expected_checksum: bytes,
        container_metadata: dict | None = None,
    ):
        self.output_path = output_path
        self.original_size = original_size
        self.expected_checksum = expected_checksum
        self.expected_n4 = level_counts(original_size)[4]
        self._container_global_dict_meta = _extract_container_global_dict_meta(container_metadata)

        self._out_f = output_path.open("wb")
        self._hasher = hashlib.blake2b(digest_size=16)
        self._out_buf = bytearray()

        self._head_size = struct.calcsize("<III")
        self._head_buf = bytearray()
        self._header_parsed = False

        self.manifest_count = 0
        self.index_count = 0
        self.packed_size = 0

        self._manifest_bytes_remaining = 0
        self._manifest_carry = b""
        self._manifest: dict[int, int] = {}
        self._max_gid = 0

        self._bit_width: int | None = None
        self._mask = 0
        self._packed_bytes_remaining = 0
        self._bitbuf = 0
        self._bits = 0

        self._produced_idx = 0
        self._produced_bytes = 0
        self._total_in = 0
        self._expected_payload_bytes: int | None = None
        self._closed = False

    def _flush_out(self) -> None:
        if self._out_buf:
            self._out_f.write(self._out_buf)
            self._hasher.update(self._out_buf)
            self._out_buf.clear()

    def _emit_label(self, label: int) -> None:
        remaining = self.original_size - self._produced_bytes
        if remaining <= 0:
            return

        b1 = (label >> 24) & 0xFF
        b2 = (label >> 16) & 0xFF
        b3 = (label >> 8) & 0xFF
        b4 = label & 0xFF

        if remaining >= 4:
            self._out_buf.extend((b1, b2, b3, b4))
            self._produced_bytes += 4
        else:
            if remaining == 3:
                self._out_buf.extend((b1, b2, b3))
            elif remaining == 2:
                self._out_buf.extend((b1, b2))
            else:
                self._out_buf.append(b1)
            self._produced_bytes += remaining

        if len(self._out_buf) >= (1 << 20):
            self._flush_out()

    def feed(self, data: bytes | memoryview) -> None:
        if self._closed:
            raise ValueError("Decoder already closed")
        if not data:
            return

        mv = memoryview(data)
        self._total_in += len(mv)
        pos = 0
        mv_len = len(mv)

        while pos < mv_len:
            if not self._header_parsed:
                need = self._head_size - len(self._head_buf)
                take = min(need, mv_len - pos)
                self._head_buf.extend(mv[pos : pos + take])
                pos += take
                if len(self._head_buf) == self._head_size:
                    self.manifest_count, self.index_count, self.packed_size = struct.unpack(
                        "<III",
                        self._head_buf,
                    )
                    if self.manifest_count == 0 and self.index_count > 0:
                        raise ValueError(
                            "Invalid global-manifest payload (empty manifest with non-empty index stream)"
                        )
                    if self.index_count != self.expected_n4:
                        raise ValueError(
                            f"Corrupted container: expected {self.expected_n4} L4 labels, got {self.index_count}"
                        )
                    self._manifest_bytes_remaining = self.manifest_count * 8
                    self._packed_bytes_remaining = self.packed_size
                    self._expected_payload_bytes = self._head_size + self._manifest_bytes_remaining + self._packed_bytes_remaining
                    self._header_parsed = True
                continue

            if self._manifest_bytes_remaining > 0:
                take = min(self._manifest_bytes_remaining, mv_len - pos)
                part = bytes(mv[pos : pos + take])
                pos += take
                self._manifest_bytes_remaining -= take

                tmp = self._manifest_carry + part
                complete = (len(tmp) // 8) * 8
                cursor = 0
                while cursor < complete:
                    gid, label = struct.unpack("<II", tmp[cursor : cursor + 8])
                    gid = int(gid)
                    if gid in self._manifest:
                        raise ValueError("Invalid global-manifest payload (duplicate gid)")
                    self._manifest[gid] = int(label)
                    if gid > self._max_gid:
                        self._max_gid = gid
                    cursor += 8
                self._manifest_carry = tmp[complete:]

                if self._manifest_bytes_remaining == 0:
                    if self._manifest_carry:
                        raise ValueError("Invalid global-manifest payload alignment")
                    self._bit_width = max(1, self._max_gid.bit_length())
                    self._mask = (1 << self._bit_width) - 1
                    expected_packed_size = (self.index_count * self._bit_width + 7) // 8
                    if self.packed_size != expected_packed_size:
                        raise ValueError("Invalid global-manifest payload (packed size mismatch)")
                continue

            if self._packed_bytes_remaining > 0:
                if self._bit_width is None:
                    raise ValueError("Invalid global-manifest payload (manifest not ready)")
                take = min(self._packed_bytes_remaining, mv_len - pos)
                part = mv[pos : pos + take]
                pos += take
                self._packed_bytes_remaining -= take

                for byte in part:
                    self._bitbuf |= int(byte) << self._bits
                    self._bits += 8
                    while self._bits >= self._bit_width and self._produced_idx < self.index_count:
                        gid = self._bitbuf & self._mask
                        self._bitbuf >>= self._bit_width
                        self._bits -= self._bit_width
                        try:
                            label = self._manifest[int(gid)]
                        except KeyError as exc:
                            raise ValueError(
                                "Corrupted global-manifest payload: gid missing from manifest"
                            ) from exc
                        self._produced_idx += 1
                        self._emit_label(label)
                continue

            raise ValueError("Invalid global-manifest payload (extra bytes)")

    def finish(self) -> int:
        if self._closed:
            raise ValueError("Decoder already closed")

        try:
            if not self._header_parsed:
                raise ValueError("Invalid global-manifest payload (missing header)")
            if self._manifest_bytes_remaining != 0:
                raise ValueError("Invalid global-manifest payload (manifest truncated)")
            if self._packed_bytes_remaining != 0:
                raise ValueError("Invalid global-manifest payload (packed stream truncated)")
            if self._produced_idx != self.index_count:
                raise ValueError(
                    f"Corrupted container: expected {self.index_count} indices, got {self._produced_idx}"
                )
            if self._produced_bytes != self.original_size:
                raise ValueError(
                    f"Corrupted container: expected {self.original_size} bytes, got {self._produced_bytes}"
                )
            if self._expected_payload_bytes is not None and self._total_in != self._expected_payload_bytes:
                raise ValueError("Invalid global-manifest payload (size mismatch)")
            if self._container_global_dict_meta is not None:
                embedded_flag = self._container_global_dict_meta.get("embedded_manifest", True)
                if embedded_flag is False:
                    raise ValueError("Container metadata mismatch: embedded manifest flag is false")
                meta_manifest_count = self._container_global_dict_meta.get("manifest_count")
                if meta_manifest_count is not None and int(meta_manifest_count) != self.manifest_count:
                    raise ValueError("Container metadata mismatch: manifest_count differs from payload")

            self._flush_out()
            self._out_f.close()
            self._closed = True
            if self._hasher.digest() != self.expected_checksum:
                raise ValueError("Checksum mismatch: decoded data is not identical to original")
            return self._produced_bytes
        except Exception:
            self.abort()
            raise

    def abort(self) -> None:
        if not self._closed:
            try:
                self._out_f.close()
            finally:
                self._closed = True
            try:
                if self.output_path.exists():
                    self.output_path.unlink()
            except Exception:
                pass


def decompress_file_stream_gdict_or_zgdict(
    input_path: Path,
    output_path: Path,
    mode_id: int,
    original_size: int,
    payload_size: int,
    checksum: bytes,
    payload_offset: int,
    chunk_size: int,
    global_dict_db: Path | None,
    container_metadata: dict | None = None,
    verbose: bool = True,
) -> dict:
    if mode_id not in (MODE_GLOBAL_DICT, MODE_ZGLOBAL_DICT, MODE_RANS_GLOBAL_DICT):
        raise ValueError("Streaming global dict decoder supports only gdict/zgdict/ansgdict modes")
    if global_dict_db is None:
        raise ValueError("gdict/zgdict/ansgdict modes require --global-dict-db")

    decoder = GlobalDictPayloadStreamingDecoder(
        output_path=output_path,
        original_size=original_size,
        expected_checksum=checksum,
        global_dict_db=Path(global_dict_db),
        container_global_dict_meta=_extract_container_global_dict_meta(container_metadata),
    )

    try:
        if mode_id == MODE_GLOBAL_DICT:
            with input_path.open("rb") as in_f:
                if payload_size >= (64 * 1024 * 1024):
                    mm = mmap.mmap(in_f.fileno(), 0, access=mmap.ACCESS_READ)
                    try:
                        start = payload_offset
                        end = payload_offset + payload_size
                        pos = start
                        while pos < end:
                            nxt = min(pos + chunk_size, end)
                            decoder.feed(memoryview(mm)[pos:nxt])
                            pos = nxt
                    finally:
                        mm.close()
                else:
                    in_f.seek(payload_offset)
                    remaining = payload_size
                    while remaining > 0:
                        chunk = in_f.read(min(chunk_size, remaining))
                        if not chunk:
                            raise ValueError("Invalid container: payload truncated")
                        remaining -= len(chunk)
                        decoder.feed(chunk)
        elif mode_id == MODE_ZGLOBAL_DICT:
            with input_path.open("rb") as in_f:
                in_f.seek(payload_offset)
                decomp = zlib.decompressobj()
                remaining = payload_size
                while remaining > 0:
                    comp_chunk = in_f.read(min(chunk_size, remaining))
                    if not comp_chunk:
                        raise ValueError("Invalid container: payload truncated")
                    remaining -= len(comp_chunk)

                    dec = decomp.decompress(comp_chunk)
                    if dec:
                        decoder.feed(dec)

                tail = decomp.flush()
                if tail:
                    decoder.feed(tail)
                if decomp.unused_data:
                    raise ValueError("Corrupted zgdict payload: trailing compressed data")
        else:
            rans_decoder = AutoRansStreamingDecoder(decoder.feed)
            with input_path.open("rb") as in_f:
                in_f.seek(payload_offset)
                remaining = payload_size
                while remaining > 0:
                    comp_chunk = in_f.read(min(chunk_size, remaining))
                    if not comp_chunk:
                        raise ValueError("Invalid container: payload truncated")
                    remaining -= len(comp_chunk)
                    rans_decoder.feed(comp_chunk)
            rans_decoder.finish()

        produced = decoder.finish()
    except Exception:
        decoder.abort()
        raise

    stats = {
        "output_size": produced,
        "mode_id": mode_id,
        "mode_name": mode_name(mode_id),
        "streaming": True,
        "global_dict_db": str(Path(global_dict_db)),
    }
    if decoder._container_global_dict_meta is not None:
        stats["global_dict_snapshot_id"] = int(decoder._container_global_dict_meta["snapshot_id"])
        stats["global_dict_db_uuid"] = str(decoder._container_global_dict_meta["db_uuid"])
    if verbose:
        print(
            f"Decompressed: {output_path} ({produced} bytes), "
            f"mode={stats['mode_name']}, stream=yes, gdict={global_dict_db}"
            + (
                f", snapshot={stats['global_dict_snapshot_id']}"
                if "global_dict_snapshot_id" in stats
                else ""
            )
        )
    return stats


def decompress_file_stream_gdictm_or_zgdictm(
    input_path: Path,
    output_path: Path,
    mode_id: int,
    original_size: int,
    payload_size: int,
    checksum: bytes,
    payload_offset: int,
    chunk_size: int,
    container_metadata: dict | None = None,
    verbose: bool = True,
) -> dict:
    if mode_id not in (
        MODE_GLOBAL_DICT_MANIFEST,
        MODE_ZGLOBAL_DICT_MANIFEST,
        MODE_RANS_GLOBAL_DICT_MANIFEST,
    ):
        raise ValueError("Streaming global-manifest decoder supports only gdictm/zgdictm/ansgdictm modes")

    decoder = GlobalDictManifestPayloadStreamingDecoder(
        output_path=output_path,
        original_size=original_size,
        expected_checksum=checksum,
        container_metadata=container_metadata,
    )

    try:
        if mode_id == MODE_GLOBAL_DICT_MANIFEST:
            with input_path.open("rb") as in_f:
                if payload_size >= (64 * 1024 * 1024):
                    mm = mmap.mmap(in_f.fileno(), 0, access=mmap.ACCESS_READ)
                    try:
                        start = payload_offset
                        end = payload_offset + payload_size
                        pos = start
                        while pos < end:
                            nxt = min(pos + chunk_size, end)
                            decoder.feed(memoryview(mm)[pos:nxt])
                            pos = nxt
                    finally:
                        mm.close()
                else:
                    in_f.seek(payload_offset)
                    remaining = payload_size
                    while remaining > 0:
                        chunk = in_f.read(min(chunk_size, remaining))
                        if not chunk:
                            raise ValueError("Invalid container: payload truncated")
                        remaining -= len(chunk)
                        decoder.feed(chunk)
        elif mode_id == MODE_ZGLOBAL_DICT_MANIFEST:
            with input_path.open("rb") as in_f:
                in_f.seek(payload_offset)
                decomp = zlib.decompressobj()
                remaining = payload_size
                while remaining > 0:
                    comp_chunk = in_f.read(min(chunk_size, remaining))
                    if not comp_chunk:
                        raise ValueError("Invalid container: payload truncated")
                    remaining -= len(comp_chunk)

                    dec = decomp.decompress(comp_chunk)
                    if dec:
                        decoder.feed(dec)

                tail = decomp.flush()
                if tail:
                    decoder.feed(tail)
                if decomp.unused_data:
                    raise ValueError("Corrupted zgdictm payload: trailing compressed data")
        else:
            rans_decoder = AutoRansStreamingDecoder(decoder.feed)
            with input_path.open("rb") as in_f:
                in_f.seek(payload_offset)
                remaining = payload_size
                while remaining > 0:
                    comp_chunk = in_f.read(min(chunk_size, remaining))
                    if not comp_chunk:
                        raise ValueError("Invalid container: payload truncated")
                    remaining -= len(comp_chunk)
                    rans_decoder.feed(comp_chunk)
            rans_decoder.finish()

        produced = decoder.finish()
    except Exception:
        decoder.abort()
        raise

    stats = {
        "output_size": produced,
        "mode_id": mode_id,
        "mode_name": mode_name(mode_id),
        "streaming": True,
        "embedded_manifest": True,
        "manifest_count": int(decoder.manifest_count),
    }
    if decoder._container_global_dict_meta is not None:
        stats["global_dict_snapshot_id"] = int(decoder._container_global_dict_meta["snapshot_id"])
        stats["global_dict_db_uuid"] = str(decoder._container_global_dict_meta["db_uuid"])
    if verbose:
        suffix = ""
        if "global_dict_snapshot_id" in stats:
            suffix = f", snapshot={stats['global_dict_snapshot_id']}"
        print(
            f"Decompressed: {output_path} ({produced} bytes), "
            f"mode={stats['mode_name']}, stream=yes, manifest={stats['manifest_count']}{suffix}"
        )
    return stats


def compress_file(
    input_path: Path,
    output_path: Path,
    mode: str = "auto",
    verbose: bool = True,
    stream: bool = False,
    chunk_size: int = 4 * 1024 * 1024,
    dict_backend: str = "sqlite",
    workers: int = 1,
    global_dict_db: Path | None = None,
) -> dict:
    source_size = input_path.stat().st_size
    metadata = build_source_metadata(input_path, source_size)

    if stream:
        if mode == "auto":
            raise ValueError(
                "Streaming compress requires explicit --mode "
                "(raw/zraw/ansraw/dict/zdict/ansdict/gdict/zgdict/ansgdict/gdictm/zgdictm/ansgdictm/ansbpegdict/ansbpegdictm)"
            )
        if mode == "ansbpe":
            raise ValueError("Streaming encoder is not implemented yet for ansbpe")
        if mode in ("raw", "zraw", "ansraw"):
            return compress_file_stream_raw_or_zraw(
                input_path,
                output_path,
                MODE_NAME_TO_ID[mode],
                chunk_size=chunk_size,
                workers=workers,
                metadata=metadata,
                verbose=verbose,
            )
        if mode in ("dict", "zdict", "ansdict"):
            if dict_backend != "sqlite":
                raise ValueError("Streaming dict/zdict/ansdict currently requires --dict-backend sqlite")
            return compress_file_stream_dict_or_zdict_sqlite(
                input_path,
                output_path,
                MODE_NAME_TO_ID[mode],
                chunk_size=chunk_size,
                workers=workers,
                metadata=metadata,
                verbose=verbose,
            )
        if mode in ("gdict", "zgdict"):
            return compress_file_stream_gdict_or_zgdict_sqlite(
                input_path,
                output_path,
                MODE_NAME_TO_ID[mode],
                chunk_size=chunk_size,
                global_dict_db=global_dict_db,
                workers=workers,
                metadata=metadata,
                verbose=verbose,
            )
        if mode in ("ansgdict",):
            return compress_file_stream_gdict_or_zgdict_sqlite(
                input_path,
                output_path,
                MODE_NAME_TO_ID[mode],
                chunk_size=chunk_size,
                global_dict_db=global_dict_db,
                workers=workers,
                metadata=metadata,
                verbose=verbose,
            )
        if mode in ("gdictm", "zgdictm"):
            return compress_file_stream_gdictm_or_zgdictm_sqlite(
                input_path,
                output_path,
                MODE_NAME_TO_ID[mode],
                chunk_size=chunk_size,
                global_dict_db=global_dict_db,
                workers=workers,
                metadata=metadata,
                verbose=verbose,
            )
        if mode in ("ansgdictm",):
            return compress_file_stream_gdictm_or_zgdictm_sqlite(
                input_path,
                output_path,
                MODE_NAME_TO_ID[mode],
                chunk_size=chunk_size,
                global_dict_db=global_dict_db,
                workers=workers,
                metadata=metadata,
                verbose=verbose,
            )
        if mode in ("ansbpegdict", "ansbpegdictm"):
            return compress_file_stream_bpe_gdict_sqlite(
                input_path,
                output_path,
                MODE_NAME_TO_ID[mode],
                chunk_size=chunk_size,
                global_dict_db=global_dict_db,
                workers=workers,
                metadata=metadata,
                verbose=verbose,
            )
        raise ValueError(f"Unsupported stream mode: {mode}")

    if mode in ("gdict", "zgdict", "ansgdict"):
        return compress_file_stream_gdict_or_zgdict_sqlite(
            input_path,
            output_path,
            MODE_NAME_TO_ID[mode],
            chunk_size=chunk_size,
            global_dict_db=global_dict_db,
            workers=workers,
            metadata=metadata,
            verbose=verbose,
        )
    if mode in ("gdictm", "zgdictm", "ansgdictm"):
        return compress_file_stream_gdictm_or_zgdictm_sqlite(
            input_path,
            output_path,
            MODE_NAME_TO_ID[mode],
            chunk_size=chunk_size,
            global_dict_db=global_dict_db,
            workers=workers,
            metadata=metadata,
            verbose=verbose,
        )
    if mode in ("ansbpegdict", "ansbpegdictm"):
        return compress_file_bpe_gdict_sqlite(
            input_path,
            output_path,
            MODE_NAME_TO_ID[mode],
            global_dict_db=global_dict_db,
            workers=workers,
            metadata=metadata,
            verbose=verbose,
        )

    data = input_path.read_bytes()
    metadata = build_source_metadata(input_path, len(data))
    blob, stats = compress_bytes(data, mode=mode, metadata=metadata, workers=workers)
    output_path.write_bytes(blob)

    if verbose:
        n0, n1, n2, n3, n4 = level_counts(len(data))
        print(f"Input:  {input_path} ({len(data)} bytes)")
        print(
            f"Output: {output_path} ({stats['output_size']} bytes), "
            f"mode={stats['mode_name']}, ratio={stats['ratio']:.3f}x"
        )
        print(f"Levels: L0={n0}, L1={n1}, L2={n2}, L3={n3}, L4={n4}")
    return stats


def decompress_file(
    input_path: Path,
    output_path: Path,
    verbose: bool = True,
    stream: bool = False,
    chunk_size: int = 4 * 1024 * 1024,
    global_dict_db: Path | None = None,
    workers: int = 1,
) -> dict:
    mode_id, original_size, payload_size, checksum, metadata, _, payload_offset = read_container_header(input_path)

    if mode_id in (MODE_GLOBAL_DICT, MODE_ZGLOBAL_DICT, MODE_RANS_GLOBAL_DICT):
        return decompress_file_stream_gdict_or_zgdict(
            input_path,
            output_path,
            mode_id=mode_id,
            original_size=original_size,
            payload_size=payload_size,
            checksum=checksum,
            payload_offset=payload_offset,
            chunk_size=chunk_size,
            global_dict_db=global_dict_db,
            container_metadata=metadata,
            verbose=verbose,
        )
    if mode_id == MODE_RANS_BPE_GLOBAL_DICT:
        if stream:
            return decompress_file_stream_bpe_gdict(
                input_path,
                output_path,
                mode_id=mode_id,
                original_size=original_size,
                payload_size=payload_size,
                checksum=checksum,
                payload_offset=payload_offset,
                global_dict_db=global_dict_db,
                container_metadata=metadata,
                verbose=verbose,
                workers=workers,
            )
        return decompress_file_bpe_gdict(
            input_path,
            output_path,
            mode_id=mode_id,
            original_size=original_size,
            payload_size=payload_size,
            checksum=checksum,
            payload_offset=payload_offset,
            global_dict_db=global_dict_db,
            container_metadata=metadata,
            verbose=verbose,
            workers=workers,
        )
    if mode_id in (
        MODE_GLOBAL_DICT_MANIFEST,
        MODE_ZGLOBAL_DICT_MANIFEST,
        MODE_RANS_GLOBAL_DICT_MANIFEST,
    ):
        return decompress_file_stream_gdictm_or_zgdictm(
            input_path,
            output_path,
            mode_id=mode_id,
            original_size=original_size,
            payload_size=payload_size,
            checksum=checksum,
            payload_offset=payload_offset,
            chunk_size=chunk_size,
            container_metadata=metadata,
            verbose=verbose,
        )
    if mode_id == MODE_RANS_BPE_GLOBAL_DICT_MANIFEST:
        if stream:
            return decompress_file_stream_bpe_gdictm(
                input_path,
                output_path,
                mode_id=mode_id,
                original_size=original_size,
                payload_size=payload_size,
                checksum=checksum,
                payload_offset=payload_offset,
                container_metadata=metadata,
                verbose=verbose,
                workers=workers,
            )
        return decompress_file_bpe_gdictm(
            input_path,
            output_path,
            mode_id=mode_id,
            original_size=original_size,
            payload_size=payload_size,
            checksum=checksum,
            payload_offset=payload_offset,
            container_metadata=metadata,
            verbose=verbose,
            workers=workers,
        )

    # Auto-stream for very large payloads
    auto_stream = mode_id in (
        MODE_RAW_L4,
        MODE_ZLIB_RAW_L4,
        MODE_RANS_RAW_L4,
        MODE_DICT_BITPACK,
        MODE_ZLIB_DICT_BITPACK,
        MODE_RANS_DICT_BITPACK,
    ) and payload_size >= (64 * 1024 * 1024)
    if stream or auto_stream:
        if mode_id in (MODE_RAW_L4, MODE_ZLIB_RAW_L4, MODE_RANS_RAW_L4):
            return decompress_file_stream_raw_or_zraw(
                input_path,
                output_path,
                mode_id=mode_id,
                original_size=original_size,
                payload_size=payload_size,
                checksum=checksum,
                payload_offset=payload_offset,
                chunk_size=chunk_size,
                verbose=verbose,
            )
        if mode_id in (MODE_DICT_BITPACK, MODE_ZLIB_DICT_BITPACK, MODE_RANS_DICT_BITPACK):
            return decompress_file_stream_dict_or_zdict(
                input_path,
                output_path,
                mode_id=mode_id,
                original_size=original_size,
                payload_size=payload_size,
                checksum=checksum,
                payload_offset=payload_offset,
                chunk_size=chunk_size,
                verbose=verbose,
            )
        if mode_id == MODE_RANS_BPE:
            raise ValueError("Streaming decompress is not implemented yet for ansbpe")
        raise ValueError("Streaming decompress unsupported for this mode")

    blob = input_path.read_bytes()
    data, stats = decompress_bytes(blob, workers=workers)
    output_path.write_bytes(data)
    if verbose:
        print(f"Decompressed: {output_path} ({len(data)} bytes), mode={stats['mode_name']}")
    return stats


def inspect_file(input_path: Path) -> None:
    mode_id, original_size, payload_size, checksum, metadata, magic, _ = read_container_header(input_path)
    n0, n1, n2, n3, n4 = level_counts(original_size)

    print(f"File: {input_path}")
    print(f"Magic: {magic.decode('ascii', errors='ignore')}")
    print(f"Mode: {mode_name(mode_id)} ({mode_id})")
    print(f"Original size: {original_size} bytes")
    print(f"Payload size: {payload_size} bytes")
    print(f"Checksum (blake2b-128): {checksum.hex()}")
    print(f"Levels: L0={n0}, L1={n1}, L2={n2}, L3={n3}, L4={n4}")
    if metadata:
        print("Metadata:")
        for k, v in metadata.items():
            print(f"  {k}: {v}")


def _random_bytes(rng: random.Random, size: int) -> bytes:
    return bytes(rng.getrandbits(8) for _ in range(size))


def _expect_failure(blob: bytes, reason: str) -> None:
    try:
        decompress_bytes(blob)
    except Exception:
        return
    raise AssertionError(f"Expected failure: {reason}")


def selftest(
    random_cases: int = 128,
    max_random_size: int = 32768,
    seed: int = 12345,
    stream_cases: int = 12,
) -> None:
    rng = random.Random(seed)

    fixed_vectors = [
        b"",
        b"\x00",
        b"\xff",
        b"\x01\x02\x03\x04\x05\x06\x07",
        bytes(range(256)),
        bytes(range(256)) * 16,
        b"A" * 1024,
        b"ABCD" * 2048,
        (b"header," + b"row1,10,20\n" * 500),
    ]

    modes = ["auto", "raw", "zraw", "dict", "zdict", "ansraw", "ansdict", "ansbpe"]
    total_ok = 0

    for i, data in enumerate(fixed_vectors):
        mode = modes[i % len(modes)]
        blob, _ = compress_bytes(data, mode=mode)
        restored, _ = decompress_bytes(blob)
        if restored != data:
            raise AssertionError(f"Fixed vector mismatch at index {i}")
        total_ok += 1

    for i in range(random_cases):
        mode = modes[i % len(modes)]
        size = rng.randint(0, max_random_size)
        variant = i % 5

        if variant == 0:
            data = _random_bytes(rng, size)
        elif variant == 1:
            b = rng.randint(0, 255)
            data = bytes([b]) * size
        elif variant == 2:
            pat = _random_bytes(rng, 8)
            data = (pat * (size // 8 + 1))[:size]
        elif variant == 3:
            data = bytes((j % 256) for j in range(size))
        else:
            arr = bytearray(size)
            flips = min(size, max(1, size // 32))
            for _ in range(flips):
                if size:
                    pos = rng.randint(0, size - 1)
                    arr[pos] = rng.randint(1, 255)
            data = bytes(arr)

        blob, _ = compress_bytes(data, mode=mode)
        restored, _ = decompress_bytes(blob)
        if restored != data:
            raise AssertionError(f"Random vector mismatch at case {i}")
        total_ok += 1

    codec_probe = (b"rans-compatibility-check-" * 64) + bytes(range(37))
    if rans_decode_auto_bytes(rans_encode_bytes(codec_probe)) != codec_probe:
        raise AssertionError("Legacy rANS compatibility decode failed")
    total_ok += 1
    if rans_decode_auto_bytes(rans_encode_framed_bytes(codec_probe, block_size=257)) != codec_probe:
        raise AssertionError("Framed rANS decode failed")
    total_ok += 1
    parallel_probe = (b"parallel-rans-check-" * 400) + bytes(range(91))
    for mode in ("ansraw", "ansdict", "ansbpe"):
        blob, _ = compress_bytes(parallel_probe, mode=mode, workers=2)
        restored, _ = decompress_bytes(blob, workers=2)
        if restored != parallel_probe:
            raise AssertionError(f"Parallel framed ANS mismatch for mode={mode}")
        total_ok += 1

    with tempfile.TemporaryDirectory(prefix="vcomp_streamtest_") as td:
        td_path = Path(td)
        global_db = td_path / "global_patterns.sqlite"
        for i in range(stream_cases):
            size = rng.randint(0, max_random_size * 8)
            if i % 3 == 0:
                data = _random_bytes(rng, size)
            elif i % 3 == 1:
                data = (b"\x11\x22\x33\x44" * (size // 4 + 1))[:size]
            else:
                data = bytes((j * 11 + i) & 0xFF for j in range(size))

            in_path = td_path / f"in_{i}.bin"
            vc_path = td_path / f"in_{i}.vcomp"
            out_path = td_path / f"out_{i}.bin"
            in_path.write_bytes(data)

            for mode in ("raw", "zraw", "ansraw"):
                compress_file(
                    in_path,
                    vc_path,
                    mode=mode,
                    verbose=False,
                    stream=True,
                    chunk_size=1024,
                    workers=2,
                )
                decompress_file(vc_path, out_path, verbose=False, stream=True, chunk_size=1024)
                if out_path.read_bytes() != data:
                    raise AssertionError(f"Streaming mismatch at case {i}, mode={mode}")
                total_ok += 1

            for mode in ("dict", "zdict", "ansdict"):
                compress_file(
                    in_path,
                    vc_path,
                    mode=mode,
                    verbose=False,
                    stream=True,
                    chunk_size=1024,
                    dict_backend="sqlite",
                    workers=2,
                )
                decompress_file(vc_path, out_path, verbose=False, stream=True, chunk_size=1024)
                if out_path.read_bytes() != data:
                    raise AssertionError(f"Streaming mismatch at case {i}, mode={mode}")
                total_ok += 1

            for mode in ("gdict", "zgdict"):
                compress_file(
                    in_path,
                    vc_path,
                    mode=mode,
                    verbose=False,
                    stream=True,
                    chunk_size=1024,
                    workers=2,
                    global_dict_db=global_db,
                )
                decompress_file(
                    vc_path,
                    out_path,
                    verbose=False,
                    stream=True,
                    chunk_size=1024,
                    global_dict_db=global_db,
                )
                if out_path.read_bytes() != data:
                    raise AssertionError(f"Global-stream mismatch at case {i}, mode={mode}")
                total_ok += 1

            for mode in ("ansgdict", "ansgdictm"):
                compress_file(
                    in_path,
                    vc_path,
                    mode=mode,
                    verbose=False,
                    stream=True,
                    chunk_size=1024,
                    workers=2,
                    global_dict_db=global_db,
                )
                decompress_file(
                    vc_path,
                    out_path,
                    verbose=False,
                    stream=True,
                    chunk_size=1024,
                    global_dict_db=(global_db if mode == "ansgdict" else None),
                )
                if out_path.read_bytes() != data:
                    raise AssertionError(f"ANS global-stream mismatch at case {i}, mode={mode}")
                total_ok += 1

            for mode in ("gdictm", "zgdictm"):
                compress_file(
                    in_path,
                    vc_path,
                    mode=mode,
                    verbose=False,
                    stream=True,
                    chunk_size=1024,
                    workers=2,
                    global_dict_db=global_db,
                )
                decompress_file(
                    vc_path,
                    out_path,
                    verbose=False,
                    stream=True,
                    chunk_size=1024,
                )
                if out_path.read_bytes() != data:
                    raise AssertionError(f"Global-manifest mismatch at case {i}, mode={mode}")
                total_ok += 1

            if i < 3:
                for mode in ("ansbpegdict", "ansbpegdictm"):
                    compress_file(
                        in_path,
                        vc_path,
                        mode=mode,
                        verbose=False,
                        stream=True,
                        chunk_size=1024,
                        workers=2,
                        global_dict_db=global_db,
                    )
                    decompress_file(
                        vc_path,
                        out_path,
                        verbose=False,
                        stream=True,
                        global_dict_db=(global_db if mode == "ansbpegdict" else None),
                        workers=2,
                    )
                    if out_path.read_bytes() != data:
                        raise AssertionError(f"BPE global mismatch at case {i}, mode={mode}")
                    total_ok += 1

        bpe_seed_a = td_path / "bpe_seed_a.bin"
        bpe_seed_b = td_path / "bpe_seed_b.bin"
        bpe_seed_vc = td_path / "bpe_seed.vcomp"
        bpe_seed_out = td_path / "bpe_seed.out"
        bpe_seed_a.write_bytes((b"ABCDABCDWXYZWXYZ" * 2048) + bytes(range(64)))
        bpe_seed_b.write_bytes((b"ABCDABCDWXYZWXYZ" * 2048) + bytes(range(64)))

        stats_a = compress_file(
            bpe_seed_a,
            bpe_seed_vc,
            mode="ansbpegdict",
            verbose=False,
            workers=2,
            global_dict_db=global_db,
        )
        total_ok += 1
        stats_b = compress_file(
            bpe_seed_b,
            bpe_seed_vc,
            mode="ansbpegdict",
            verbose=False,
            workers=2,
            global_dict_db=global_db,
        )
        if int(stats_b.get("bpe_seed_pairs_available", 0)) <= 0:
            raise AssertionError("Persistent BPE cache did not expose seed pairs on second encode")
        if int(stats_b.get("bpe_seed_merges_applied", 0)) <= 0:
            raise AssertionError("Persistent BPE cache did not apply seed merges on second encode")
        if int(stats_b.get("bpe_shared_vocab_pairs_available", 0)) <= 0:
            raise AssertionError("Persistent BPE vocabulary did not expose ranked pairs on second encode")
        if int(stats_b.get("bpe_shared_vocab_merges_applied", 0)) <= 0:
            raise AssertionError("Persistent BPE vocabulary did not apply ranked merges on second encode")
        decompress_file(
            bpe_seed_vc,
            bpe_seed_out,
            verbose=False,
            global_dict_db=global_db,
            workers=2,
        )
        if bpe_seed_out.read_bytes() != bpe_seed_b.read_bytes():
            raise AssertionError("Persistent BPE cache roundtrip failed")
        total_ok += 2

        snapshot_data = (b"snapshot-check-" * 700) + bytes(range(32))
        snapshot_in = td_path / "snapshot_in.bin"
        snapshot_vc = td_path / "snapshot.vcomp"
        snapshot_out = td_path / "snapshot_out.bin"
        snapshot_growth_in = td_path / "snapshot_growth.bin"
        other_global_db = td_path / "global_patterns_other.sqlite"
        other_vc = td_path / "other_snapshot.vcomp"

        snapshot_in.write_bytes(snapshot_data)
        compress_file(
            snapshot_in,
            snapshot_vc,
            mode="zgdict",
            verbose=False,
            stream=True,
            chunk_size=1024,
            workers=2,
            global_dict_db=global_db,
        )

        snapshot_growth_in.write_bytes(bytes((j * 17 + 3) & 0xFF for j in range(8192)))
        compress_file(
            snapshot_growth_in,
            td_path / "snapshot_growth.vcomp",
            mode="gdict",
            verbose=False,
            stream=True,
            chunk_size=1024,
            workers=2,
            global_dict_db=global_db,
        )
        decompress_file(
            snapshot_vc,
            snapshot_out,
            verbose=False,
            stream=True,
            chunk_size=1024,
            global_dict_db=global_db,
        )
        if snapshot_out.read_bytes() != snapshot_data:
            raise AssertionError("Snapshot validation failed after append-only DB growth")
        total_ok += 1

        compress_file(
            snapshot_in,
            other_vc,
            mode="gdict",
            verbose=False,
            stream=True,
            chunk_size=1024,
            workers=1,
            global_dict_db=other_global_db,
        )
        try:
            decompress_file(
                snapshot_vc,
                snapshot_out,
                verbose=False,
                stream=True,
                chunk_size=1024,
                global_dict_db=other_global_db,
            )
        except Exception:
            pass
        else:
            raise AssertionError("Expected global dictionary snapshot mismatch on wrong DB")

        compress_file(
            snapshot_in,
            td_path / "snapshot_manifest.vcomp",
            mode="zgdictm",
            verbose=False,
            stream=True,
            chunk_size=1024,
            workers=2,
            global_dict_db=global_db,
        )
        decompress_file(
            td_path / "snapshot_manifest.vcomp",
            snapshot_out,
            verbose=False,
            stream=True,
            chunk_size=1024,
        )
        if snapshot_out.read_bytes() != snapshot_data:
            raise AssertionError("Embedded manifest decode failed without external global DB")
        total_ok += 1

    sample = (b"vcomp-corruption-check-" * 200) + bytes(range(64))
    good_blob, _ = compress_bytes(sample, mode="zdict")

    bad_magic = bytearray(good_blob)
    bad_magic[0] ^= 0xFF
    _expect_failure(bytes(bad_magic), "bad magic")

    bad_mode = bytearray(good_blob)
    bad_mode[8] = 255
    _expect_failure(bytes(bad_mode), "bad mode")

    truncated = good_blob[:-1]
    _expect_failure(truncated, "truncated payload")

    _, _, _, _, _, parsed_magic, _ = parse_container_blob(good_blob)
    checksum_offset = CHECKSUM_OFFSET_V2 if parsed_magic == MAGIC_V2 else CHECKSUM_OFFSET_V1
    header_size = HEADER_V2_SIZE if parsed_magic == MAGIC_V2 else HEADER_V1_SIZE

    bad_checksum = bytearray(good_blob)
    bad_checksum[checksum_offset] ^= 0xFF
    _expect_failure(bytes(bad_checksum), "checksum mismatch")

    bad_payload = bytearray(good_blob)
    if len(bad_payload) > header_size:
        bad_payload[-1] ^= 0xFF
    _expect_failure(bytes(bad_payload), "payload corruption")

    print(f"Self-test: OK ({total_ok} roundtrips + corruption tests)")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="VCOMP prototype (multilevel lossless compressor)."
    )
    sub = parser.add_subparsers(dest="cmd", required=True)

    p_c = sub.add_parser("compress", help="Compress file to .vcomp")
    p_c.add_argument("input", type=Path)
    p_c.add_argument("output", type=Path)
    p_c.add_argument(
        "--mode",
        choices=[
            "auto",
            "raw",
            "zraw",
            "dict",
            "zdict",
            "ansraw",
            "ansdict",
            "ansbpe",
            "ansbpegdict",
            "ansbpegdictm",
            "ansgdict",
            "ansgdictm",
            "gdict",
            "zgdict",
            "gdictm",
            "zgdictm",
        ],
        default="auto",
        help="Payload mode. Default: auto",
    )
    p_c.add_argument(
        "--stream",
        action="store_true",
        help="Use streaming encoder (requires explicit mode)",
    )
    p_c.add_argument("--chunk-mib", type=int, default=4, help="Chunk size in MiB for stream mode")
    p_c.add_argument(
        "--dict-backend",
        choices=["memory", "sqlite"],
        default="sqlite",
        help="Dictionary backend for stream dict/zdict/ansdict modes (default: sqlite)",
    )
    p_c.add_argument(
        "--workers",
        type=int,
        default=1,
        help="Worker processes for stream modes and non-stream framed ANS modes, default: 1",
    )
    p_c.add_argument(
        "--global-dict-db",
        type=Path,
        help="Path to persistent SQLite global dictionary (required for gdict/zgdict/gdictm/zgdictm/ansgdict/ansgdictm/ansbpegdict/ansbpegdictm encode)",
    )

    p_d = sub.add_parser("decompress", help="Decompress .vcomp to original bytes")
    p_d.add_argument("input", type=Path)
    p_d.add_argument("output", type=Path)
    p_d.add_argument("--stream", action="store_true", help="Use streaming decoder")
    p_d.add_argument("--chunk-mib", type=int, default=4, help="Chunk size in MiB for stream mode")
    p_d.add_argument(
        "--workers",
        type=int,
        default=1,
        help="Worker processes for non-stream framed ANS decode, default: 1",
    )
    p_d.add_argument(
        "--global-dict-db",
        type=Path,
        help="Path to persistent SQLite global dictionary (required only for gdict/zgdict/ansgdict/ansbpegdict decode)",
    )

    p_i = sub.add_parser("info", help="Show container header and derived level stats")
    p_i.add_argument("input", type=Path)

    p_t = sub.add_parser("selftest", help="Run roundtrip and corruption tests")
    p_t.add_argument("--random-cases", type=int, default=128)
    p_t.add_argument("--max-random-size", type=int, default=32768)
    p_t.add_argument("--seed", type=int, default=12345)
    p_t.add_argument("--stream-cases", type=int, default=12)

    return parser.parse_args()


def main() -> int:
    args = parse_args()
    try:
        if args.cmd == "compress":
            compress_file(
                args.input,
                args.output,
                mode=args.mode,
                stream=args.stream,
                chunk_size=max(1, args.chunk_mib) * 1024 * 1024,
                dict_backend=args.dict_backend,
                workers=max(1, args.workers),
                global_dict_db=args.global_dict_db,
            )
        elif args.cmd == "decompress":
            decompress_file(
                args.input,
                args.output,
                stream=args.stream,
                chunk_size=max(1, args.chunk_mib) * 1024 * 1024,
                global_dict_db=args.global_dict_db,
                workers=max(1, args.workers),
            )
        elif args.cmd == "info":
            inspect_file(args.input)
        elif args.cmd == "selftest":
            selftest(
                random_cases=args.random_cases,
                max_random_size=args.max_random_size,
                seed=args.seed,
                stream_cases=args.stream_cases,
            )
        else:
            raise ValueError(f"Unknown command: {args.cmd}")
        return 0
    except Exception as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())