Upload 17 files

__init__.py

@@ -21,9 +21,18 @@ from .factored import (
     factor_only_solve,
     factor_q2k_block,
 )
-from .gguf import GGUFReader, expand_importance, lookup_imatrix, read_imatrix
+from .gguf import (
+    GGUFReader,
+    expand_importance,
+    is_q2k_eligible,
+    is_q4_0_eligible,
+    lookup_imatrix,
+    read_imatrix,
+    target_quant_type,
+)
 from .pipeline import Q2KQuantizer, QuantizedTensor, TensorStats
 from .q2k import Q2KBlock, QK_K, dequantize_params, pack_codes, unpack_codes
+from .q4_0 import QK4_0, dequantize_q4_0_bytes, quantize_tensor_q4_0
 
 __all__ = [
     "HPCConfig",
@@ -47,8 +56,14 @@ __all__ = [
     "factor_q2k_block",
     "GGUFReader",
     "expand_importance",
+    "is_q2k_eligible",
+    "is_q4_0_eligible",
     "lookup_imatrix",
     "read_imatrix",
+    "target_quant_type",
+    "QK4_0",
+    "dequantize_q4_0_bytes",
+    "quantize_tensor_q4_0",
     "pack_codes",
     "unpack_codes",
 ]

__main__.py

@@ -26,6 +26,7 @@ from .gguf import (
     is_q2k_eligible,
     lookup_imatrix,
     read_imatrix,
+    target_quant_type,
     write_quantized_gguf,
 )
 from .pipeline import Q2KQuantizer
@@ -46,6 +47,16 @@ def _config_from_args(args) -> HPCConfig:
         cfg.optimizer = "mse"
     if args.no_spectral:
         cfg.use_spectral = False
+    if args.no_q4_fallback:
+        cfg.enable_q4_fallback = False
+    if args.keep_embd:
+        cfg.quantize_embd_q4 = False
+    if args.promote_attn_q4:
+        cfg.promote_attn_q4 = True
+    if args.format is not None:
+        cfg.quant_format = args.format
+        if args.format == "q2_k":
+            cfg.enable_q4_fallback = False
     return cfg.validate()
 
 
@@ -82,9 +93,18 @@ def _run_gguf(args, cfg: HPCConfig) -> None:
     if args.list_tensors or (not args.tensor and not args.all_q2k):
         print(f"{reader.path}: {len(reader.tensors)} tensors, {len(reader.metadata)} metadata keys")
         for ti in reader.tensors:
-            tag = "Q2K" if is_q2k_eligible(ti.name, ti.dims) else "   "
+            tgt = target_quant_type(
+                ti.name,
+                ti.dims,
+                enable_q4_fallback=cfg.enable_q4_fallback,
+                quantize_embd_q4=cfg.quantize_embd_q4,
+                promote_attn_q4=cfg.promote_attn_q4,
+                quant_format=cfg.quant_format,
+                source_type=ti.type,
+            )
+            tag = tgt if tgt else "COPY"
             print(
-                f"  {tag} {ti.name:48s} {ti.type_name:6s} "
+                f"  {tag:4s} {ti.name:48s} {ti.type_name:6s} "
                 f"dims={ti.dims} row={ti.row_width}"
             )
         return
@@ -108,8 +128,8 @@ def _run_gguf(args, cfg: HPCConfig) -> None:
             verbose=True,
         )
         print(
-            f"wrote {args.output}: {int(stats['quantized'])}/{int(stats['tensors'])} tensors Q2_K, "
-            f"rmse={stats['rmse']:.6e}"
+            f"wrote {args.output}: {int(stats.get('q2_k', 0))} Q2_K + {int(stats.get('q4_0', 0))} Q4_0 "
+            f"({int(stats['quantized'])}/{int(stats['tensors'])} tensors), rmse={stats['rmse']:.6e}"
         )
         return
 
@@ -231,6 +251,15 @@ def main() -> None:
     parser.add_argument("--device", default=None, help="compute device: auto (default), cuda, or cpu")
     parser.add_argument("--mse", action="store_true", help="disable HPC sequence search")
     parser.add_argument("--no-spectral", action="store_true", help="disable spectral/DC objective")
+    parser.add_argument(
+        "--format",
+        choices=("mixed", "q4_0", "q2_k"),
+        default=None,
+        help="quantization format mode: mixed (Q2_K if eligible else Q4_0), q4_0 (pure Q4_0), or q2_k (pure Q2_K)",
+    )
+    parser.add_argument("--no-q4-fallback", action="store_true", help="disable Q4_0 fallback for non-256-aligned tensors")
+    parser.add_argument("--keep-embd", action="store_true", help="keep token_embd/output.weight at source precision")
+    parser.add_argument("--promote-attn-q4", action="store_true", help="promote attention Q/K/V/O projections to Q4_0")
     parser.add_argument("--verbose", action="store_true")
     parser.add_argument("--bench", action="store_true", help="measure solver gains vs C port")
     parser.add_argument("--bench-per-kind", type=int, default=12)

config.py

@@ -68,6 +68,10 @@ class HPCConfig:
     use_factor_floor: bool = True
     use_candidate_floor: bool = True
     apply_dc_carry: bool = True
+    quant_format: str = "mixed"
+    enable_q4_fallback: bool = True
+    quantize_embd_q4: bool = True
+    promote_attn_q4: bool = False
 
     # ---- seed clamp -----------------------------------------------------
     clamp_kurtosis_off: float = 6.0
@@ -116,4 +120,6 @@ class HPCConfig:
             raise ValueError("workers must be >= 0 (0 = auto CPU count)")
         if self.device not in ("auto", "cuda", "cpu") and not self.device.startswith("cuda:"):
             raise ValueError("device must be 'auto', 'cuda', 'cuda:N', or 'cpu'")
+        if self.quant_format not in ("mixed", "q4_0", "q2_k"):
+            raise ValueError("quant_format must be 'mixed', 'q4_0', or 'q2_k'")
         return self

gguf.py

@@ -22,6 +22,7 @@ from pathlib import Path
 import numpy as np
 
 from .q2k import QK_K, blocks_to_bytes, dequantize_params_batch, unpack_codes
+from .q4_0 import Q4_0_BLOCK_BYTES, QK4_0, quantize_tensor_q4_0
 
 GGUF_MAGIC = 0x46554747
 DEFAULT_ALIGNMENT = 32
@@ -70,6 +71,38 @@ BLOCK_BYTES = {
     GGML_TYPE_Q8_0: 34,
     GGML_TYPE_Q2_K: 84,
 }
+TYPE_BPW = {
+    GGML_TYPE_F32: 32.0,
+    GGML_TYPE_F16: 16.0,
+    GGML_TYPE_BF16: 16.0,
+    GGML_TYPE_Q8_0: 8.5,
+    GGML_TYPE_Q5_1: 6.0,
+    GGML_TYPE_Q5_0: 5.5,
+    GGML_TYPE_Q4_1: 5.0,
+    GGML_TYPE_Q4_0: 4.5,
+    GGML_TYPE_Q2_K: 2.625,
+    11: 3.4375,  # Q3_K
+    12: 4.5,     # Q4_K
+    13: 5.5,     # Q5_K
+    14: 6.5625,  # Q6_K
+    15: 8.5,     # Q8_K
+    16: 2.0625,  # IQ2_XXS
+    17: 2.3125,  # IQ2_XS
+    18: 3.0625,  # IQ3_XXS
+    19: 1.5625,  # IQ1_S
+    20: 4.5,     # IQ4_NL
+    21: 3.4375,  # IQ3_S
+    22: 2.5625,  # IQ2_S
+    23: 4.25,    # IQ4_XS
+}
+
+
+def source_type_bpw(ttype: int) -> float:
+    if ttype in TYPE_BPW:
+        return TYPE_BPW[ttype]
+    if ttype in BLOCK_SIZE and ttype in BLOCK_BYTES:
+        return (BLOCK_BYTES[ttype] * 8.0) / BLOCK_SIZE[ttype]
+    return 16.0
 SUPPORTED_READ_TYPES = {
     GGML_TYPE_F32,
     GGML_TYPE_F16,
@@ -229,6 +262,30 @@ class GGUFReader:
         raw = self.read_tensor_raw(name)
         return _dequantize(raw, self.info(name)), self.info(name).row_width
 
+    def iter_tensor_row_chunks(self, name: str, max_elements: int = 8_388_608):
+        """Stream a tensor in row-aligned float32 chunks to bound peak RAM."""
+        ti = self.info(name)
+        row = ti.row_width
+        blk = BLOCK_SIZE[ti.type]
+        if row <= 0 or row % blk != 0 or ti.n_elements % row != 0:
+            x, r = self.read_tensor(name)
+            yield x, r, 0, ti.n_elements // max(r, 1)
+            return
+        n_rows = ti.n_elements // row
+        rows_per_chunk = max(1, max_elements // row)
+        bytes_per_row = (row // blk) * BLOCK_BYTES[ti.type]
+        with self.path.open("rb") as f:
+            for r0 in range(0, n_rows, rows_per_chunk):
+                r1 = min(r0 + rows_per_chunk, n_rows)
+                chunk_elems = (r1 - r0) * row
+                chunk_bytes = (r1 - r0) * bytes_per_row
+                f.seek(self.data_start + ti.offset + r0 * bytes_per_row)
+                raw = f.read(chunk_bytes)
+                if len(raw) != chunk_bytes:
+                    raise ValueError(f"{name}: truncated tensor data")
+                sub_ti = TensorInfo(ti.name, (chunk_elems,), ti.type, 0)
+                yield _dequantize(raw, sub_ti), row, r0, r1
+
 
 def _dequantize(raw: bytes, ti: TensorInfo) -> np.ndarray:
     t = ti.type
@@ -343,24 +400,36 @@ def expand_importance(
     )
 
 
-def is_q2k_eligible(name: str, dims: tuple[int, ...]) -> bool:
+def _is_skipped_metadata_tensor(name: str, dims: tuple[int, ...]) -> bool:
     if len(dims) < 2:
-        return False
-    if dims[0] % QK_K != 0:
-        return False
-    if "token_embd" in name or name == "output.weight":
-        return False
+        return True
     if "norm" in name or ".bias" in name:
-        return False
+        return True
     if "ffn_gate_inp" in name:
-        return False
+        return True
     if name.startswith("mm.") or name.startswith("a.") or name.startswith("v."):
-        return False
+        return True
     if "altup" in name or "laurel" in name or "layer_output_scale" in name:
+        return True
+    if "ssm_a" in name.split(".") or "A_log" in name:
+        return True
+    if "ssm_dt" in name or "dt_bias" in name:
+        return True
+    if "ssm_conv1d" in name or "conv1d.weight" in name:
+        return True
+    return False
+
+
+def _is_embd_or_head(name: str) -> bool:
+    return "token_embd" in name or name == "output.weight" or name.endswith(".output.weight")
+
+
+def is_q2k_eligible(name: str, dims: tuple[int, ...]) -> bool:
+    if _is_skipped_metadata_tensor(name, dims):
         return False
-    if "ssm_a" in name or "ssm_dt" in name or "dt_bias" in name:
+    if _is_embd_or_head(name):
         return False
-    if "ssm_conv1d" in name:
+    if dims[0] % QK_K != 0:
         return False
     n_elements = 1
     for d in dims:
@@ -368,6 +437,79 @@ def is_q2k_eligible(name: str, dims: tuple[int, ...]) -> bool:
     return n_elements >= QK_K and n_elements % QK_K == 0
 
 
+def is_q4_0_eligible(
+    name: str,
+    dims: tuple[int, ...],
+    include_embd: bool = True,
+    promote_attn: bool = False,
+    allow_q2k_tensors: bool = False,
+) -> bool:
+    """True for 32-aligned 2D/3D weight tensors eligible for Q4_0."""
+    if _is_skipped_metadata_tensor(name, dims):
+        return False
+    is_embd = _is_embd_or_head(name)
+    if is_embd and not include_embd:
+        return False
+    if not allow_q2k_tensors and not is_embd and not (promote_attn and _is_attention_projection(name)):
+        if is_q2k_eligible(name, dims):
+            return False
+    if dims[0] % QK4_0 != 0:
+        return False
+    n_elements = 1
+    for d in dims:
+        n_elements *= int(d)
+    return n_elements >= QK4_0 and n_elements % QK4_0 == 0
+
+
+def target_quant_type(
+    name: str,
+    dims: tuple[int, ...],
+    enable_q4_fallback: bool = True,
+    quantize_embd_q4: bool = True,
+    promote_attn_q4: bool = False,
+    quant_format: str = "mixed",
+    source_type: int | None = None,
+) -> str | None:
+    """Determine target quantization format ('Q2_K', 'Q4_0', or None for Copy).
+
+    Modes:
+    * 'mixed' (default): all Q2_K-eligible tensors -> Q2_K; non-Q2_K 32-aligned
+      tensors (and embeddings/head if quantize_embd_q4) -> Q4_0.
+    * 'q4_0': all 32-aligned eligible weight tensors -> Q4_0.
+    * 'q2_k': only Q2_K-eligible tensors -> Q2_K; everything else copied.
+
+    Never promotes a lower-bit source tensor to a higher-bit target format
+    (e.g., a Q2_K source tensor is never re-quantized to Q4_0).
+    """
+    tgt: str | None = None
+    if quant_format == "q4_0":
+        if is_q4_0_eligible(
+            name, dims, include_embd=quantize_embd_q4, allow_q2k_tensors=True
+        ):
+            tgt = "Q4_0"
+    elif quant_format == "q2_k":
+        if is_q2k_eligible(name, dims):
+            tgt = "Q2_K"
+    else:
+        if promote_attn_q4 and _is_attention_projection(name) and is_q4_0_eligible(
+            name, dims, include_embd=quantize_embd_q4, promote_attn=True
+        ):
+            tgt = "Q4_0"
+        elif is_q2k_eligible(name, dims):
+            tgt = "Q2_K"
+        elif enable_q4_fallback and is_q4_0_eligible(
+            name, dims, include_embd=quantize_embd_q4, promote_attn=False
+        ):
+            tgt = "Q4_0"
+
+    if tgt is not None and source_type is not None:
+        src_bpw = source_type_bpw(int(source_type))
+        tgt_bpw = 2.625 if tgt == "Q2_K" else 4.5
+        if src_bpw <= tgt_bpw + 1.0e-6:
+            return None
+    return tgt
+
+
 def _write_string(f, value: str) -> None:
     data = value.encode("utf-8")
     f.write(struct.pack("<Q", len(data)))
@@ -400,52 +542,83 @@ def write_quantized_gguf(
         if missing:
             raise KeyError(f"tensors not found: {sorted(missing)}")
 
-    plan: list[bool] = []
+    cfg = getattr(quantizer, "config", None)
+    enable_q4 = bool(getattr(cfg, "enable_q4_fallback", True))
+    embd_q4 = bool(getattr(cfg, "quantize_embd_q4", True))
+    attn_q4 = bool(getattr(cfg, "promote_attn_q4", False))
+    q_format = str(getattr(cfg, "quant_format", "mixed"))
+
+    plan: list[str | None] = []
     quantized_count = 0
     for ti in reader.tensors:
+        target_fmt: str | None = None
         if names is not None:
-            want = ti.name in names
-            compatible = (
-                len(ti.dims) >= 2
-                and ti.dims[0] % QK_K == 0
-                and ti.n_elements % QK_K == 0
+            if ti.name in names:
+                target_fmt = target_quant_type(
+                    ti.name,
+                    ti.dims,
+                    enable_q4_fallback=enable_q4,
+                    quantize_embd_q4=embd_q4,
+                    promote_attn_q4=attn_q4,
+                    quant_format=q_format,
+                    source_type=ti.type,
+                )
+                if target_fmt is None:
+                    src_bpw = source_type_bpw(ti.type)
+                    if q_format == "q4_0" and len(ti.dims) >= 2 and ti.dims[0] % QK4_0 == 0 and src_bpw > 4.5:
+                        target_fmt = "Q4_0"
+                    elif len(ti.dims) >= 2 and ti.dims[0] % QK_K == 0 and ti.n_elements % QK_K == 0 and src_bpw > 2.625:
+                        target_fmt = "Q2_K"
+                    elif len(ti.dims) >= 2 and ti.dims[0] % QK4_0 == 0 and ti.n_elements % QK4_0 == 0 and src_bpw > 4.5:
+                        target_fmt = "Q4_0"
+                    elif verbose:
+                        print(f"  copy (already <= target precision or unaligned): {ti.name}")
+        elif quantize_all:
+            target_fmt = target_quant_type(
+                ti.name,
+                ti.dims,
+                enable_q4_fallback=enable_q4,
+                quantize_embd_q4=embd_q4,
+                promote_attn_q4=attn_q4,
+                quant_format=q_format,
+                source_type=ti.type,
             )
-        else:
-            want = quantize_all and is_q2k_eligible(ti.name, ti.dims)
-            compatible = True
-        if want and ti.type not in SUPPORTED_READ_TYPES:
+        if target_fmt and ti.type not in SUPPORTED_READ_TYPES:
             if verbose:
                 print(f"  copy (unsupported source {ti.type_name}): {ti.name}")
-            want = False
-        if want and not compatible:
-            if verbose:
-                print(f"  copy (row width not Q2_K aligned): {ti.name}")
-            want = False
-        if want and limit and quantized_count >= limit:
-            want = False
-        if want:
+            target_fmt = None
+        if target_fmt and limit and quantized_count >= limit:
+            target_fmt = None
+        if target_fmt:
             quantized_count += 1
-        plan.append(want)
+        plan.append(target_fmt)
 
     out_infos = []
     data_offset = 0
     total_planned_blocks = 0
-    for ti, want in zip(reader.tensors, plan):
-        if want:
+    for ti, fmt in zip(reader.tensors, plan):
+        if fmt == "Q2_K":
             n_blocks = ti.n_elements // QK_K
             out_type = GGML_TYPE_Q2_K
             out_size = n_blocks * 84
             total_planned_blocks += n_blocks
+        elif fmt == "Q4_0":
+            n_blocks = ti.n_elements // QK4_0
+            out_type = GGML_TYPE_Q4_0
+            out_size = n_blocks * Q4_0_BLOCK_BYTES
+            total_planned_blocks += max(1, n_blocks // 8)
         else:
             out_type = ti.type
             out_size = ti.n_bytes
         out_infos.append((ti, out_type, out_size, data_offset))
         data_offset = _align(data_offset + out_size, alignment)
 
+    has_q2k = any(fmt == "Q2_K" for fmt in plan)
+    ftype_code = 10 if has_q2k else 2  # 10 = MOSTLY_Q2_K, 2 = MOSTLY_Q4_0
     kv_pairs = []
     for key, vtype, raw in reader.kv_pairs:
         if quantized_count and key == "general.file_type" and vtype == 4:
-            raw = struct.pack("<I", 10)
+            raw = struct.pack("<I", ftype_code)
         elif quantized_count and key == "general.quantization_version" and vtype == 4:
             raw = struct.pack("<I", 2)
         kv_pairs.append((key, vtype, raw))
@@ -454,6 +627,8 @@ def write_quantized_gguf(
     output_path.parent.mkdir(parents=True, exist_ok=True)
     total_rmse = 0.0
     total_quantized = 0
+    total_q2k = 0
+    total_q4_0 = 0
     total_elements = 0
     completed_planned_blocks = 0
     model_t0 = time.time()
@@ -480,7 +655,8 @@ def write_quantized_gguf(
             target = data_start + offset
             if target > fout.tell():
                 fout.write(b"\x00" * (target - fout.tell()))
-            if plan[idx]:
+            fmt = plan[idx]
+            if fmt == "Q2_K":
                 t_blocks = ti.n_elements // QK_K
                 has_imat = imat is not None and lookup_imatrix(imat, ti.name) is not None
 
@@ -490,11 +666,11 @@ def write_quantized_gguf(
                     phase_name = ev.get("phase", "")
                     done_b = int(ev.get("done", 0))
                     tot_b = max(int(ev.get("total", _tb)), 1)
-                    if phase_name.startswith("Phase 2b"):
+                    if phase_name.startswith("Phase 2b") or "Phase 1–2b" in phase_name:
                         frac = 0.80 * (done_b / tot_b)
                     elif phase_name.startswith("Phase 3"):
                         frac = 0.82
-                    elif phase_name.startswith("Phase 4:"):
+                    elif "Phase 4" in phase_name and not phase_name.startswith("Phase 4.8"):
                         frac = 0.82 + 0.17 * (done_b / tot_b)
                     elif phase_name.startswith("Phase 4.8") or phase_name.startswith("Phase 5"):
                         frac = 1.0
@@ -548,6 +724,7 @@ def write_quantized_gguf(
                 total_rmse += res.stats.rmse * res.stats.rmse * x.size
                 total_elements += x.size
                 total_quantized += 1
+                total_q2k += 1
                 if verbose:
                     print(
                         f"  Q2_K {ti.name:48s} row={row:6d} "
@@ -562,6 +739,78 @@ def write_quantized_gguf(
                         float(res.stats.rmse),
                         imp is not None,
                     )
+            elif fmt == "Q4_0":
+                t_blocks_q4 = ti.n_elements // QK4_0
+                eff_tb = max(1, t_blocks_q4 // 8)
+                raw_imp = lookup_imatrix(imat, ti.name) if imat is not None else None
+                has_imat = raw_imp is not None and raw_imp.size == ti.row_width
+                written_bytes = 0
+                sse_q4 = 0.0
+                done_q4_blks = 0
+                t_q4_start = time.time()
+
+                for x_chunk, row, _r0, _r1 in reader.iter_tensor_row_chunks(ti.name, max_elements=8_388_608):
+                    if stop_event is not None and stop_event.is_set():
+                        raise InterruptedError("Quantization cancelled by user")
+                    imp_chunk = expand_importance(raw_imp, x_chunk.size, row) if has_imat else None
+                    q4_bytes, nb_chunk, chunk_rmse = quantize_tensor_q4_0(
+                        x_chunk, importance=imp_chunk, cfg=cfg
+                    )
+                    fout.write(q4_bytes)
+                    written_bytes += len(q4_bytes)
+                    sse_q4 += (chunk_rmse * chunk_rmse) * x_chunk.size
+                    done_q4_blks += nb_chunk
+                    if stage_callback is not None:
+                        frac = done_q4_blks / max(t_blocks_q4, 1)
+                        eff_blocks = completed_planned_blocks + frac * eff_tb
+                        elapsed_model = max(time.time() - model_t0, 1.0e-6)
+                        model_rate = eff_blocks / elapsed_model
+                        model_eta = max(total_planned_blocks - eff_blocks, 0.0) / max(model_rate, 1.0e-6)
+                        el_q4 = max(time.time() - t_q4_start, 1.0e-6)
+                        r_q4 = done_q4_blks / el_q4
+                        eta_q4 = (t_blocks_q4 - done_q4_blks) / max(r_q4, 1.0e-6)
+                        stage_callback(
+                            idx + 1,
+                            len(out_infos),
+                            ti,
+                            {
+                                "phase": "Q4_0: Analog + WLS + FP16-ULP Quantization",
+                                "done": done_q4_blks,
+                                "total": t_blocks_q4,
+                                "detail": f"blk {done_q4_blks:,}/{t_blocks_q4:,} ({r_q4:.0f} blk/s) rmse={chunk_rmse:.6e}",
+                                "rate": r_q4,
+                                "eta": eta_q4,
+                                "rmse": chunk_rmse,
+                                "sigma": float(np.sqrt(np.mean(x_chunk.astype(np.float64) ** 2))),
+                                "model_pct": 100.0 * eff_blocks / max(total_planned_blocks, 1),
+                                "model_eta": model_eta,
+                                "model_done_blocks": int(eff_blocks),
+                                "model_total_blocks": total_planned_blocks,
+                            },
+                        )
+
+                if written_bytes != out_size:
+                    raise RuntimeError(f"{ti.name}: Q4_0 payload size mismatch")
+                completed_planned_blocks += eff_tb
+                q4_rmse = float(np.sqrt(sse_q4 / ti.n_elements))
+                total_rmse += sse_q4
+                total_elements += ti.n_elements
+                total_quantized += 1
+                total_q4_0 += 1
+                if verbose:
+                    print(
+                        f"  Q4_0 {ti.name:48s} row={ti.row_width:6d} "
+                        f"rmse={q4_rmse:.6e} imatrix={'yes' if has_imat else 'no'}"
+                    )
+                if progress_callback is not None:
+                    progress_callback(
+                        idx + 1,
+                        len(out_infos),
+                        ti.name,
+                        "Q4_0",
+                        q4_rmse,
+                        has_imat,
+                    )
             else:
                 if stage_callback is not None:
                     stage_callback(
@@ -585,6 +834,8 @@ def write_quantized_gguf(
     return {
         "tensors": float(len(out_infos)),
         "quantized": float(total_quantized),
+        "q2_k": float(total_q2k),
+        "q4_0": float(total_q4_0),
         "elements": float(total_elements),
         "rmse": rmse,
     }

gui.py

@@ -41,7 +41,7 @@ from hpc_quantize_v2 import (
     lookup_imatrix,
     read_imatrix,
 )
-from hpc_quantize_v2.gguf import is_q2k_eligible, write_quantized_gguf
+from hpc_quantize_v2.gguf import is_q2k_eligible, target_quant_type, write_quantized_gguf
 from hpc_quantize_v2.reference import reference_solve, synthetic_cases
 
 
@@ -110,6 +110,11 @@ class HPCQuantizeGUI:
         self.factor_floor_var = tk.BooleanVar(value=True)
         self.cand_floor_var = tk.BooleanVar(value=True)
         self.dc_carry_var = tk.BooleanVar(value=True)
+        self.q4_fallback_var = tk.BooleanVar(value=True)
+        self.embd_q4_var = tk.BooleanVar(value=True)
+        self.promote_attn_q4_var = tk.BooleanVar(value=False)
+        self.quant_format_var = tk.StringVar(value="mixed")
+        self.plan_filter_var = tk.StringVar(value="ALL")
 
         self.dc_lambda_var = tk.DoubleVar(value=1.0)
         self.vw_lambda_var = tk.DoubleVar(value=1.0)
@@ -143,6 +148,10 @@ class HPCQuantizeGUI:
             self.workers_var,
             self.device_var,
             self.use_spectral_var,
+            self.q4_fallback_var,
+            self.embd_q4_var,
+            self.promote_attn_q4_var,
+            self.quant_format_var,
         ):
             v.trace_add("write", lambda *_: self._update_cli_preview())
 
@@ -178,8 +187,48 @@ class HPCQuantizeGUI:
         style.map("Treeview", background=[("selected", accent)])
 
     def _build_layout(self) -> None:
-        main = ttk.Frame(self.root, padding=10)
-        main.pack(fill=tk.BOTH, expand=True)
+        outer = ttk.Frame(self.root)
+        outer.pack(fill=tk.BOTH, expand=True)
+        outer.rowconfigure(0, weight=1)
+        outer.columnconfigure(0, weight=1)
+
+        self._canvas = tk.Canvas(outer, bg="#141820", highlightthickness=0)
+        v_scroll = ttk.Scrollbar(outer, orient=tk.VERTICAL, command=self._canvas.yview)
+        h_scroll = ttk.Scrollbar(outer, orient=tk.HORIZONTAL, command=self._canvas.xview)
+        self._canvas.configure(yscrollcommand=v_scroll.set, xscrollcommand=h_scroll.set)
+
+        self._canvas.grid(row=0, column=0, sticky=tk.NSEW)
+        v_scroll.grid(row=0, column=1, sticky=tk.NS)
+        h_scroll.grid(row=1, column=0, sticky=tk.EW)
+
+        main = ttk.Frame(self._canvas, padding=10)
+        self._canvas_win = self._canvas.create_window((0, 0), window=main, anchor=tk.NW)
+
+        def _on_frame_configure(_event=None) -> None:
+            self._canvas.configure(scrollregion=self._canvas.bbox("all"))
+
+        def _on_canvas_configure(event) -> None:
+            req_w = max(event.width, main.winfo_reqwidth())
+            req_h = max(event.height, main.winfo_reqheight())
+            self._canvas.itemconfigure(self._canvas_win, width=req_w, height=req_h)
+
+        main.bind("<Configure>", _on_frame_configure)
+        self._canvas.bind("<Configure>", _on_canvas_configure)
+
+        def _on_mousewheel(event) -> None:
+            w = event.widget
+            if isinstance(w, (ttk.Treeview, tk.Text)):
+                return
+            if getattr(event, "num", None) == 4:
+                self._canvas.yview_scroll(-1, "units")
+            elif getattr(event, "num", None) == 5:
+                self._canvas.yview_scroll(1, "units")
+            elif getattr(event, "delta", 0):
+                self._canvas.yview_scroll(int(-1 * (event.delta / 120)), "units")
+
+        self.root.bind_all("<MouseWheel>", _on_mousewheel)
+        self.root.bind_all("<Button-4>", _on_mousewheel)
+        self.root.bind_all("<Button-5>", _on_mousewheel)
 
         # 1. File Selection Frame
         files_frame = ttk.LabelFrame(main, text="1. GGUF & iMatrix Paths", padding=8)
@@ -217,12 +266,28 @@ class HPCQuantizeGUI:
         mode_row.pack(fill=tk.X, pady=(0, 4))
         ttk.Label(mode_row, text="Target Scope:").pack(side=tk.LEFT, padx=(0, 8))
         ttk.Radiobutton(
-            mode_row, text="All Q2_K-Eligible Tensors", variable=self.mode_var, value="all"
+            mode_row, text="All Eligible Tensors", variable=self.mode_var, value="all"
         ).pack(side=tk.LEFT, padx=4)
         ttk.Radiobutton(
             mode_row, text="Selected Tensor(s)", variable=self.mode_var, value="selected"
         ).pack(side=tk.LEFT, padx=4)
 
+        fmt_row = ttk.Frame(opt_frame)
+        fmt_row.pack(fill=tk.X, pady=(0, 4))
+        ttk.Label(fmt_row, text="Quant Format:").pack(side=tk.LEFT, padx=(0, 8))
+        for lbl, val in (
+            ("Mixed (Q2_K if eligible, else Q4_0)", "mixed"),
+            ("Pure Q4_0", "q4_0"),
+            ("Pure Q2_K", "q2_k"),
+        ):
+            ttk.Radiobutton(
+                fmt_row,
+                text=lbl,
+                variable=self.quant_format_var,
+                value=val,
+                command=self._populate_tensor_tree,
+            ).pack(side=tk.LEFT, padx=3)
+
         opt_row = ttk.Frame(opt_frame)
         opt_row.pack(fill=tk.X, pady=(0, 4))
         ttk.Label(opt_row, text="Global Optimizer:").pack(side=tk.LEFT, padx=(0, 8))
@@ -265,6 +330,24 @@ class HPCQuantizeGUI:
             text="Local-Safety Candidate & Factored Floors (Monotone Bound)",
             variable=self.factor_floor_var,
         ).pack(anchor=tk.W, pady=1)
+        ttk.Checkbutton(
+            checks,
+            text="Q4_0 Fallback for 32-Aligned Non-Q2_K Tensors & Recurrent Gates (ssm_alpha/beta, ffn_down_exps, hc_*)",
+            variable=self.q4_fallback_var,
+            command=self._populate_tensor_tree,
+        ).pack(anchor=tk.W, pady=1)
+        ttk.Checkbutton(
+            checks,
+            text="Quantize token_embd.weight & output.weight to Q4_0 (Saves 1.3–3.7 GB)",
+            variable=self.embd_q4_var,
+            command=self._populate_tensor_tree,
+        ).pack(anchor=tk.W, pady=1)
+        ttk.Checkbutton(
+            checks,
+            text="Promote Attention Q/K/V/O Projections to Q4_0 (Mixed Q2_K + Q4_0)",
+            variable=self.promote_attn_q4_var,
+            command=self._populate_tensor_tree,
+        ).pack(anchor=tk.W, pady=1)
 
         grid_f = ttk.Frame(opt_frame)
         grid_f.pack(fill=tk.X, pady=4)
@@ -303,23 +386,34 @@ class HPCQuantizeGUI:
 
         search_row = ttk.Frame(browser_frame)
         search_row.pack(fill=tk.X, pady=(0, 6))
-        ttk.Label(search_row, text="Filter:").pack(side=tk.LEFT, padx=(0, 6))
+        ttk.Label(search_row, text="Filter:").pack(side=tk.LEFT, padx=(0, 4))
         filter_entry = ttk.Entry(search_row, textvariable=self.filter_var)
         filter_entry.pack(side=tk.LEFT, fill=tk.X, expand=True)
         filter_entry.bind("<KeyRelease>", lambda _e: self._populate_tensor_tree())
+        for label, val in (("All", "ALL"), ("Q2_K", "Q2_K"), ("Q4_0", "Q4_0"), ("Copy", "COPY")):
+            ttk.Radiobutton(
+                search_row,
+                text=label,
+                variable=self.plan_filter_var,
+                value=val,
+                command=self._populate_tensor_tree,
+            ).pack(side=tk.LEFT, padx=3)
         ttk.Button(
-            search_row, text="Select Eligible", command=self._select_all_eligible
+            search_row, text="Select Quantized", command=self._select_all_eligible
         ).pack(side=tk.LEFT, padx=4)
 
         cols = ("name", "type", "dims", "row", "q2k", "imat", "state")
         self.tree = ttk.Treeview(browser_frame, columns=cols, show="headings", selectmode="extended")
         self.tree.heading("name", text="Tensor Name")
-        self.tree.heading("type", text="Type")
+        self.tree.heading("type", text="Source")
         self.tree.heading("dims", text="Dims")
         self.tree.heading("row", text="Row Width")
-        self.tree.heading("q2k", text="Q2_K")
+        self.tree.heading("q2k", text="Target")
         self.tree.heading("imat", text="iMatrix")
         self.tree.heading("state", text="Live Status / RMSE")
+        self.tree.tag_configure("Q2_K", foreground="#93c5fd")
+        self.tree.tag_configure("Q4_0", foreground="#34d399")
+        self.tree.tag_configure("Copy", foreground="#9ca3af")
 
         self.tree.column("name", width=220, anchor=tk.W)
         self.tree.column("type", width=52, anchor=tk.CENTER)
@@ -497,6 +591,10 @@ class HPCQuantizeGUI:
             vw_lambda=float(self.vw_lambda_var.get()),
             dc_decay=float(self.dc_decay_var.get()),
             sse_budget=float(self.sse_budget_var.get()),
+            quant_format=self.quant_format_var.get(),
+            enable_q4_fallback=bool(self.q4_fallback_var.get()) and self.quant_format_var.get() != "q2_k",
+            quantize_embd_q4=bool(self.embd_q4_var.get()),
+            promote_attn_q4=bool(self.promote_attn_q4_var.get()),
             device=self.device_var.get(),
             workers=int(self.workers_var.get()),
             verbose=False,
@@ -521,6 +619,7 @@ class HPCQuantizeGUI:
             cmd += ["--output", out]
         if self.limit_var.get() > 0:
             cmd += ["--limit", str(self.limit_var.get())]
+        cmd += ["--format", self.quant_format_var.get()]
         cmd += ["--device", self.device_var.get()]
         cmd += ["--workers", str(self.workers_var.get())]
         cmd += ["--cand-d", str(self.cand_d_var.get()), "--cand-m", str(self.cand_m_var.get())]
@@ -528,6 +627,12 @@ class HPCQuantizeGUI:
             cmd.append("--mse")
         if not self.use_spectral_var.get():
             cmd.append("--no-spectral")
+        if not self.q4_fallback_var.get():
+            cmd.append("--no-q4-fallback")
+        if not self.embd_q4_var.get():
+            cmd.append("--keep-embd")
+        if self.promote_attn_q4_var.get():
+            cmd.append("--promote-attn-q4")
         cmd.append("--verbose")
         return " ".join(shlex.quote(part) for part in cmd)
 
@@ -560,12 +665,6 @@ class HPCQuantizeGUI:
             return
         try:
             self._reader = GGUFReader(gguf_path)
-            eligible = sum(1 for t in self._reader.tensors if is_q2k_eligible(t.name, t.dims))
-            self._log(
-                f"Inspected {gguf_path}: {len(self._reader.tensors)} tensors "
-                f"({eligible} Q2_K-eligible), {len(self._reader.metadata)} metadata keys."
-            )
-            self.status_var.set(f"Loaded {len(self._reader.tensors)} tensors ({eligible} Q2_K-eligible).")
             self._populate_tensor_tree()
         except Exception as exc:
             self._reader = None
@@ -576,28 +675,62 @@ class HPCQuantizeGUI:
         if self._reader is None:
             return
         filt = self.filter_var.get().strip().lower()
+        plan_filt = self.plan_filter_var.get()
+        n_q2, n_q4, n_copy = 0, 0, 0
         for ti in self._reader.tensors:
-            if filt and filt not in ti.name.lower() and filt not in ti.type_name.lower():
+            tgt = target_quant_type(
+                ti.name,
+                ti.dims,
+                enable_q4_fallback=bool(self.q4_fallback_var.get()) and self.quant_format_var.get() != "q2_k",
+                quantize_embd_q4=bool(self.embd_q4_var.get()),
+                promote_attn_q4=bool(self.promote_attn_q4_var.get()),
+                quant_format=self.quant_format_var.get(),
+                source_type=ti.type,
+            )
+            if tgt == "Q2_K":
+                n_q2 += 1
+            elif tgt == "Q4_0":
+                n_q4 += 1
+            else:
+                n_copy += 1
+
+            tgt_label = tgt if tgt else "Copy"
+            if plan_filt != "ALL" and tgt_label.upper() != plan_filt:
+                continue
+            if (
+                filt
+                and filt not in ti.name.lower()
+                and filt not in ti.type_name.lower()
+                and filt not in tgt_label.lower()
+            ):
                 continue
-            elig = is_q2k_eligible(ti.name, ti.dims)
             has_imat = (
                 self._imatrix is not None
                 and lookup_imatrix(self._imatrix, ti.name) is not None
             )
+            state_str = (
+                f"Pending ({tgt_label})"
+                if tgt
+                else (f"Keep {ti.type_name} (<= target)" if ti.type_name in ("Q2_K", "Q4_0") else "Passthrough")
+            )
             self.tree.insert(
                 "",
                 tk.END,
                 iid=ti.name,
+                tags=(tgt_label,),
                 values=(
                     ti.name,
                     ti.type_name,
                     str(ti.dims),
                     ti.row_width,
-                    "Yes" if elig else "No",
+                    tgt_label,
                     "Yes" if has_imat else "-",
-                    "Pending (Q2_K)" if elig else "Passthrough",
+                    state_str,
                 ),
             )
+        self.status_var.set(
+            f"Plan ({self.quant_format_var.get().upper()}): {n_q2} Q2_K | {n_q4} Q4_0 | {n_copy} Copy (Total: {len(self._reader.tensors)} tensors)"
+        )
 
     def _select_all_eligible(self) -> None:
         if self._reader is None:
@@ -605,7 +738,17 @@ class HPCQuantizeGUI:
         eligible_ids = [
             ti.name
             for ti in self._reader.tensors
-            if is_q2k_eligible(ti.name, ti.dims) and self.tree.exists(ti.name)
+            if target_quant_type(
+                ti.name,
+                ti.dims,
+                enable_q4_fallback=bool(self.q4_fallback_var.get()) and self.quant_format_var.get() != "q2_k",
+                quantize_embd_q4=bool(self.embd_q4_var.get()),
+                promote_attn_q4=bool(self.promote_attn_q4_var.get()),
+                quant_format=self.quant_format_var.get(),
+                source_type=ti.type,
+            )
+            is not None
+            and self.tree.exists(ti.name)
         ]
         self.tree.selection_set(eligible_ids)
 
@@ -737,7 +880,8 @@ class HPCQuantizeGUI:
             self._log_queue.put(
                 (
                     "done",
-                    f"Wrote {out_path}: {int(stats['quantized'])}/{int(stats['tensors'])} tensors Q2_K, "
+                    f"Wrote {out_path}: {int(stats.get('q2_k', 0))} Q2_K + {int(stats.get('q4_0', 0))} Q4_0 "
+                    f"({int(stats['quantized'])}/{int(stats['tensors'])} tensors), "
                     f"overall RMSE={stats['rmse']:.6e} ({dt_sec:.1f}s)",
                 )
             )
@@ -880,10 +1024,10 @@ class HPCQuantizeGUI:
                 done, total, name, action, rmse, has_imat = payload  # type: ignore[misc]
                 self.model_progress_var.set(100.0 * done / max(total, 1))
                 self.tensor_progress_var.set(100.0)
-                if action == "Q2_K" and rmse is not None:
-                    self._update_tree_row_state(name, f"Q2_K (RMSE={rmse:.4e})")
+                if action in ("Q2_K", "Q4_0") and rmse is not None:
+                    self._update_tree_row_state(name, f"{action} (RMSE={rmse:.4e})")
                     self._log(
-                        f"[{done}/{total}] FINISHED {name:42s} -> Q2_K RMSE={rmse:.6e} (iMatrix={'yes' if has_imat else 'no'})"
+                        f"[{done}/{total}] FINISHED {name:42s} -> {action} RMSE={rmse:.6e} (iMatrix={'yes' if has_imat else 'no'})"
                     )
                 else:
                     self._update_tree_row_state(name, "Copied verbatim")

q4_0.py

@@ -0,0 +1,444 @@
+"""Full HPC V2 Q4_0 quantizer (CUDA GPU + Vectorized CPU).
+
+Implements the complete HPC-Quantize V2 decision stack for ``Q4_0`` (32 weights
+per 18-byte block = 4.5 bpw):
+
+1. **Geometry-Preserving Analog + WLS Seeds**:
+   Finds optimal ``d*`` from both the raw weights and an iso-moment companded
+   analog of each 32-weight block.
+2. **24-Candidate Scale Lattice & Extended Spectral Objective**:
+   Evaluates 24 scale multipliers (mapped to 6 symbolic quhit states) on
+   importance-weighted SSE + D6/Vesica fold pyramid + DC penalty, including
+   negative-scale sign-flip candidates ``-d`` (which mirror the asymmetric
+   ``[-8, +7]`` grid to ``[-7, +8]`` when positive outliers dominate).
+3. **6-State Sequential Sieve & 6-State Viterbi DP**:
+   Runs the 6-state sequential Sieve with parity back-action and a 6-state
+   Viterbi dynamic program with free-energy bin reduction and Manhattan
+   transition penalty across the block chain.
+4. **Asymmetric Sign-Flip fp16-ULP Polish & Closed-Form Vesica/DC Shaping**:
+   Polishes ``d`` over ``+/-8`` fp16 ULPs of both ``+d`` and ``-d``, and
+   applies closed-form 1-D quadratic Vesica + DC code projection bounded by
+   ``sse_budget`` plus a strict monotone candidate floor.
+"""
+
+from __future__ import annotations
+
+from math import ceil
+import numpy as np
+
+try:
+    import torch
+except ImportError:  # pragma: no cover
+    torch = None  # type: ignore[assignment]
+
+from .config import HPCConfig
+from .cuda_backend import cuda_is_available
+from .factored import CAND_TO_QUHIT, MULTS_D
+from .fp16 import from_fp16_bits, nearest_int, to_fp16_bits
+from .sieve import sieve_measure_chain, subblock_chain_edges
+from .viterbi import robust_temperature
+
+QK4_0 = 32
+Q4_0_BLOCK_BYTES = 18
+Q4_N_CAND = 24
+Q4_N_STATES = 6
+
+
+def dequantize_q4_0_bytes(raw: bytes | np.ndarray, n_blocks: int) -> np.ndarray:
+    blocks = np.frombuffer(raw, dtype=np.uint8).reshape(n_blocks, Q4_0_BLOCK_BYTES)
+    d = blocks[:, :2].copy().view("<f2").reshape(n_blocks, 1).astype(np.float32)
+    qs = blocks[:, 2:]
+    low = (qs & 0x0F).astype(np.float32) - 8.0
+    high = (qs >> 4).astype(np.float32) - 8.0
+    q = np.concatenate([low, high], axis=1)
+    return (d * q).reshape(-1)
+
+
+def _q4_sieve_and_viterbi(cand_errors: np.ndarray, cfg: HPCConfig) -> np.ndarray:
+    """6-state sequential Sieve + 6-state Viterbi DP over Q4_0 candidate lattice."""
+    B = cand_errors.shape[0]
+    if cfg.optimizer != "hpc" or B < 2:
+        return np.argmin(cand_errors, axis=1).astype(np.int64)
+
+    max_gb = min(int(cfg.graph_max_blocks), 512)
+    stride = max(1, ceil(B / max_gb))
+    gb = ceil(B / stride)
+    temp = max(float(robust_temperature(cand_errors, cfg)), 1.0e-12)
+
+    pad = gb * stride - B
+    ce = cand_errors.astype(np.float64)
+    if pad > 0:
+        ce = np.concatenate([ce, np.full((pad, Q4_N_CAND), np.inf, dtype=np.float64)], axis=0)
+    ce_grouped = ce.reshape(gb, stride, Q4_N_STATES, 4)
+
+    group_min = np.min(ce_grouped, axis=(1, 3))  # (gb, 6)
+    finite = np.isfinite(ce_grouped)
+    counts = np.sum(finite, axis=(1, 3))
+    valid = (counts > 0) & np.isfinite(group_min)
+    gm_b = np.where(np.isfinite(group_min), group_min, 0.0)[:, None, :, None]
+    shifted = np.where(finite, (ce_grouped - gm_b) / (2.0 * temp), np.inf)
+    sums = np.sum(np.exp(-shifted), axis=(1, 3))
+    mean_exp = np.maximum(sums / np.maximum(counts, 1), 1.0e-300)
+    bin_err = np.where(valid, group_min - 2.0 * temp * np.log(mean_exp), np.inf)
+
+    # Sieve local amplitudes from mean candidate errors per group:
+    agg = np.mean(np.where(finite, ce_grouped, 0.0), axis=1).reshape(gb, Q4_N_CAND)
+    mn = np.min(agg, axis=1, keepdims=True)
+    coeff = np.exp(-np.clip((agg - mn) / (2.0 * temp), 0.0, 80.0)).reshape(gb, Q4_N_STATES, 4)
+    amps = np.sum(coeff, axis=2)
+    norms = np.linalg.norm(amps, axis=1, keepdims=True)
+    amps = np.where(norms > 1.0e-30, amps / np.maximum(norms, 1.0e-30), 1.0 / np.sqrt(6.0))
+
+    marg, _ = sieve_measure_chain(
+        amps,
+        subblock_chain_edges(gb),
+        same_penalty=cfg.same_penalty,
+        log_slack=cfg.log_slack,
+        deterministic=True,
+    )
+    m_tot = np.sum(marg, axis=1, keepdims=True)
+    pm = np.where(m_tot > 1.0e-30, marg / np.maximum(m_tot, 1.0e-30), 1.0 / 6.0)
+    log_prior = np.log(pm + 1.0e-30)
+
+    fin_err = bin_err[np.isfinite(bin_err)]
+    scale_err = max(float(fin_err.mean()) if fin_err.size else 1.0e-10, 1.0e-20)
+    local = np.where(np.isfinite(bin_err), bin_err - cfg.viterbi_beta * scale_err * log_prior, np.inf)
+
+    states = np.arange(Q4_N_STATES)
+    trans = cfg.viterbi_alpha * scale_err * np.abs(states[None, :] - states[:, None])
+    dp = np.full((gb, Q4_N_STATES), np.inf, dtype=np.float64)
+    back = np.zeros((gb, Q4_N_STATES), dtype=np.int64)
+    dp[0] = local[0]
+    for gi in range(1, gb):
+        totals = dp[gi - 1][:, None] + trans
+        best_sp = np.argmin(totals, axis=0)
+        best_val = totals[best_sp, states]
+        dp[gi] = np.where(np.isfinite(best_val), best_val + local[gi], np.inf)
+        back[gi] = best_sp
+
+    path = np.zeros(gb, dtype=np.int64)
+    if np.any(np.isfinite(dp[-1])):
+        path[-1] = int(np.argmin(dp[-1]))
+        for gi in range(gb - 2, -1, -1):
+            path[gi] = back[gi + 1, path[gi + 1]]
+
+    ce_bins = cand_errors.reshape(B, Q4_N_STATES, 4)
+    bin_local_arg = np.argmin(ce_bins, axis=2)  # (B, 6)
+    blk_gi = np.arange(B) // stride
+    chosen_bin = path[blk_gi]
+    chosen_cand = chosen_bin * 4 + bin_local_arg[np.arange(B), chosen_bin]
+
+    global_best = np.argmin(cand_errors, axis=1).astype(np.int64)
+    cur_e = cand_errors[np.arange(B), chosen_cand]
+    g_e = cand_errors[np.arange(B), global_best]
+    return np.where(g_e < cur_e * cfg.greedy_override_ratio, global_best, chosen_cand)
+
+
+def _quantize_q4_0_hpc_cpu(
+    X: np.ndarray, W: np.ndarray, cfg: HPCConfig
+) -> tuple[bytes, float]:
+    """Full HPC V2 Q4_0 pipeline on CPU for a chunk of (B, 32) blocks."""
+    B = X.shape[0]
+    X_f32 = X.astype(np.float32)
+    W_f32 = W.astype(np.float32)
+    lam_dc = (float(cfg.dc_lambda) / QK4_0) if cfg.use_spectral else 0.0
+    lam_vw = (float(cfg.vw_lambda) / QK4_0) if cfg.use_spectral else 0.0
+    half = QK4_0 // 2
+
+    # 1. Analog + WLS seed for d* (testing both +d and asymmetric sign-flip -d):
+    amax = np.max(np.abs(X_f32), axis=1)
+    d_range = amax / 7.5
+    wsum = np.maximum(np.sum(W_f32, axis=1, keepdims=True), 1.0e-30)
+    mu = np.sum(W_f32 * X_f32, axis=1, keepdims=True) / wsum
+    var = np.sum(W_f32 * (X_f32 - mu) ** 2, axis=1, keepdims=True) / wsum
+    sig = np.sqrt(np.maximum(var, 1.0e-30))
+    x_comp = mu + sig * 1.85 * np.tanh((X_f32 - mu) / (1.85 * sig))
+    d_analog = np.max(np.abs(x_comp), axis=1).astype(np.float32) / 7.5
+
+    wls_d = d_range.copy()
+    best_seed_sse = np.full(B, np.inf, dtype=np.float64)
+    for d_start in (d_range, -d_range, d_analog, -d_analog):
+        d = d_start.astype(np.float64)
+        for _ in range(4):
+            act = np.abs(d) > 1.0e-15
+            sd = np.where(act, d, 1.0)[:, None]
+            q = np.where(act[:, None], np.clip(nearest_int((X_f32 / sd).astype(np.float32)) + 8, 0, 15), 8)
+            c = (q - 8).astype(np.float64)
+            num = np.sum(W_f32 * X_f32 * c, axis=1, dtype=np.float64)
+            den = np.sum(W_f32 * c * c, axis=1, dtype=np.float64)
+            valid = den > 1.0e-30
+            d = np.where(valid, num / np.where(valid, den, 1.0), 0.0)
+        df = from_fp16_bits(to_fp16_bits(d.astype(np.float32)))
+        act = np.abs(df) > 1.0e-15
+        sdf = np.where(act, df, 1.0)[:, None]
+        q = np.where(act[:, None], np.clip(nearest_int(X_f32 / sdf) + 8, 0, 15), 8)
+        e = X_f32 - df[:, None] * (q.astype(np.float32) - 8.0)
+        sse = np.sum(W_f32 * e * e, axis=1, dtype=np.float64)
+        take = sse < best_seed_sse
+        best_seed_sse = np.where(take, sse, best_seed_sse)
+        wls_d = np.where(take, df, wls_d)
+
+    # 2. 24-Candidate Field with Extended Spectral Objective:
+    cand_d16 = to_fp16_bits(wls_d[:, None] * MULTS_D[None, :])  # (B, 24)
+    cand_d = from_fp16_bits(cand_d16)                           # (B, 24)
+    act_c = np.abs(cand_d) > 1.0e-15
+    sd_c = np.where(act_c, cand_d, 1.0)[:, :, None]
+    q_c = np.where(
+        act_c[:, :, None],
+        np.clip(nearest_int(X_f32[:, None, :] / sd_c) + 8, 0, 15),
+        8,
+    ).astype(np.float32)
+    e_c = X_f32[:, None, :] - cand_d[:, :, None] * (q_c - 8.0)  # (B, 24, 32)
+    sse_c = np.sum(W_f32[:, None, :] * e_c * e_c, axis=2)
+    if cfg.use_spectral and (lam_dc > 0.0 or lam_vw > 0.0):
+        v_c = e_c[:, :, :half] + e_c[:, :, half:]
+        dc_c = np.sum(e_c, axis=2)
+        cand_errors = (sse_c + lam_vw * np.sum(v_c * v_c, axis=2) + lam_dc * dc_c * dc_c).astype(np.float32)
+    else:
+        cand_errors = sse_c.astype(np.float32)
+
+    # 3. 6-State Sieve + 6-State Viterbi DP:
+    best_cand = _q4_sieve_and_viterbi(cand_errors, cfg)
+    rows = np.arange(B)
+    d_sel = cand_d[rows, best_cand]
+
+    # 4. Asymmetric Sign-Flip fp16-ULP Polish (+/-8 ULPs around +d_sel and -d_sel):
+    rad = max(int(cfg.ulp_radius), 4)
+    deltas = np.arange(-rad, rad + 1, dtype=np.int32)
+    pos_b = np.clip(to_fp16_bits(np.abs(d_sel)).astype(np.int32)[:, None] + deltas[None, :], 0, 0x7BFF).astype(np.uint16)
+    neg_b = (pos_b | np.uint16(0x8000)).astype(np.uint16)
+    all_b = np.concatenate([pos_b, neg_b], axis=1)  # (B, 2*(2*rad+1))
+    all_d = from_fp16_bits(all_b)
+    act_u = np.abs(all_d) > 1.0e-15
+    sd_u = np.where(act_u, all_d, 1.0)[:, :, None]
+    q_u = np.where(
+        act_u[:, :, None],
+        np.clip(nearest_int(X_f32[:, None, :] / sd_u) + 8, 0, 15),
+        8,
+    ).astype(np.uint8)
+    e_u = X_f32[:, None, :] - all_d[:, :, None] * (q_u.astype(np.float32) - 8.0)
+    sse_u = np.sum(W_f32[:, None, :] * e_u * e_u, axis=2, dtype=np.float64)
+    pick_u = np.argmin(sse_u, axis=1)
+    best_d16 = all_b[rows, pick_u]
+    best_d = all_d[rows, pick_u]
+    best_q = q_u[rows, pick_u].astype(np.int64)
+
+    # 5. Closed-Form 1-D Quadratic Vesica + DC Code Projection within sse_budget:
+    if cfg.use_spectral and (lam_vw > 0.0 or lam_dc > 0.0):
+        c_f = (best_q - 8).astype(np.float64)
+        e = X_f32.astype(np.float64) - best_d[:, None].astype(np.float64) * c_f
+        sse_live = np.sum(W_f32 * e * e, axis=1, dtype=np.float64)
+        sse_cap = sse_live * (1.0 + cfg.sse_budget)
+        v = e[:, :half] + e[:, half:]
+        dc = np.sum(e, axis=1)
+        d_f64 = best_d.astype(np.float64)[:, None]
+        denom = (W_f32 + lam_vw + lam_dc) * (d_f64 * d_f64)
+        act_b = np.abs(d_f64) > 1.0e-15
+        s_den = np.where(act_b & (denom > 1.0e-30), denom, 1.0)
+        v_full = np.concatenate([v, v], axis=1)
+        grad = d_f64 * (W_f32 * e + lam_vw * v_full + lam_dc * dc[:, None])
+        dq = np.where(act_b, nearest_int((grad / s_den).astype(np.float32)), 0)
+        q_new = np.clip(best_q + dq, 0, 15)
+        e_new = X_f32.astype(np.float64) - d_f64 * (q_new - 8).astype(np.float64)
+        sse_new = np.sum(W_f32 * e_new * e_new, axis=1, dtype=np.float64)
+        v_new = e_new[:, :half] + e_new[:, half:]
+        dc_new = np.sum(e_new, axis=1)
+        spec_old = lam_vw * np.sum(v * v, axis=1) + lam_dc * dc * dc
+        spec_new = lam_vw * np.sum(v_new * v_new, axis=1) + lam_dc * dc_new * dc_new
+        accept = (sse_new <= sse_cap) & ((sse_new + spec_new) < (sse_live + spec_old) - 1.0e-15)
+        best_q = np.where(accept[:, None], q_new, best_q)
+
+    best_q_u8 = best_q.astype(np.uint8)
+    out = np.empty((B, Q4_0_BLOCK_BYTES), dtype=np.uint8)
+    out[:, :2] = best_d16.astype("<u2").view(np.uint8).reshape(B, 2)
+    out[:, 2:] = (best_q_u8[:, :16] & 0x0F) | ((best_q_u8[:, 16:] & 0x0F) << np.uint8(4))
+
+    deq_final = best_d[:, None] * (best_q_u8.astype(np.float32) - 8.0)
+    raw_sse = float(np.sum((X_f32 - deq_final) ** 2, dtype=np.float64))
+    return out.tobytes(), raw_sse
+
+
+def _quantize_q4_0_hpc_cuda(
+    X_np: np.ndarray, W_np: np.ndarray, dev: "torch.device", cfg: HPCConfig
+) -> tuple[bytes, float]:
+    """Full HPC V2 Q4_0 pipeline on CUDA GPU for a chunk of (B, 32) blocks."""
+    B = X_np.shape[0]
+    lam_dc = (float(cfg.dc_lambda) / QK4_0) if cfg.use_spectral else 0.0
+    lam_vw = (float(cfg.vw_lambda) / QK4_0) if cfg.use_spectral else 0.0
+    half = QK4_0 // 2
+
+    with torch.no_grad():
+        X = torch.from_numpy(X_np).to(dev, dtype=torch.float32)
+        W = torch.from_numpy(W_np).to(dev, dtype=torch.float32)
+
+        # 1. Analog + WLS seed (including negative-scale sign-flip starts):
+        amax = torch.max(torch.abs(X), dim=1).values
+        d_range = amax / 7.5
+        wsum = torch.clamp(torch.sum(W, dim=1, keepdim=True), min=1.0e-30)
+        mu = torch.sum(W * X, dim=1, keepdim=True) / wsum
+        var = torch.sum(W * (X - mu) ** 2, dim=1, keepdim=True) / wsum
+        sig = torch.sqrt(torch.clamp(var, min=1.0e-30))
+        x_comp = mu + sig * 1.85 * torch.tanh((X - mu) / (1.85 * sig))
+        d_analog = torch.max(torch.abs(x_comp), dim=1).values / 7.5
+
+        wls_d = d_range.clone()
+        best_seed_sse = torch.full((B,), float("inf"), dtype=torch.float32, device=dev)
+        for d_start in (d_range, -d_range, d_analog, -d_analog):
+            d = d_start.clone()
+            for _ in range(4):
+                act = torch.abs(d) > 1.0e-15
+                sd = torch.where(act, d, torch.ones_like(d))[:, None]
+                sh = (X / sd) + 12582912.0
+                qi = ((sh.view(torch.int32) & 0x007FFFFF) - 0x00400000 + 8).clamp(0, 15)
+                q = torch.where(act[:, None], qi, torch.full_like(qi, 8))
+                c = (q - 8).to(torch.float32)
+                num = torch.sum(W * X * c, dim=1)
+                den = torch.sum(W * c * c, dim=1)
+                valid = den > 1.0e-30
+                d = torch.where(valid, num / torch.where(valid, den, torch.ones_like(den)), torch.zeros_like(den))
+            df = d.to(torch.float16).to(torch.float32)
+            act = torch.abs(df) > 1.0e-15
+            sdf = torch.where(act, df, torch.ones_like(df))[:, None]
+            sh = (X / sdf) + 12582912.0
+            qi = ((sh.view(torch.int32) & 0x007FFFFF) - 0x00400000 + 8).clamp(0, 15)
+            q = torch.where(act[:, None], qi, torch.full_like(qi, 8))
+            e = X - df[:, None] * (q.to(torch.float32) - 8.0)
+            sse = torch.sum(W * e * e, dim=1)
+            take = sse < best_seed_sse
+            best_seed_sse = torch.where(take, sse, best_seed_sse)
+            wls_d = torch.where(take, df, wls_d)
+
+        # 2. 24-Candidate Field with Extended Spectral Objective on GPU:
+        mults = torch.tensor(MULTS_D, dtype=torch.float32, device=dev)
+        cand_d = (wls_d[:, None] * mults[None, :]).to(torch.float16).to(torch.float32)  # (B, 24)
+        act_c = torch.abs(cand_d) > 1.0e-15
+        sd_c = torch.where(act_c, cand_d, torch.ones_like(cand_d))[:, :, None]
+        sh_c = (X[:, None, :] / sd_c) + 12582912.0
+        qi_c = ((sh_c.view(torch.int32) & 0x007FFFFF) - 0x00400000 + 8).clamp(0, 15)
+        q_c = torch.where(act_c[:, :, None], qi_c, torch.full_like(qi_c, 8)).to(torch.float32)
+        e_c = X[:, None, :] - cand_d[:, :, None] * (q_c - 8.0)
+        sse_c = torch.sum(W[:, None, :] * e_c * e_c, dim=2)
+        if cfg.use_spectral and (lam_dc > 0.0 or lam_vw > 0.0):
+            v_c = e_c[:, :, :half] + e_c[:, :, half:]
+            dc_c = torch.sum(e_c, dim=2)
+            cand_err_t = sse_c + lam_vw * torch.sum(v_c * v_c, dim=2) + lam_dc * dc_c * dc_c
+        else:
+            cand_err_t = sse_c
+
+        # 3. 6-State Sieve + 6-State Viterbi DP:
+        best_cand_np = _q4_sieve_and_viterbi(cand_err_t.cpu().numpy(), cfg)
+        best_cand_t = torch.from_numpy(best_cand_np).to(dev)
+        rows = torch.arange(B, device=dev)
+        d_sel = cand_d[rows, best_cand_t]
+
+        # 4. Asymmetric Sign-Flip fp16-ULP Polish (+/-8 ULPs around +d_sel and -d_sel):
+        rad = max(int(cfg.ulp_radius), 4)
+        deltas = torch.arange(-rad, rad + 1, dtype=torch.int32, device=dev)
+        abs_bits = torch.abs(d_sel).to(torch.float16).view(torch.int16).to(torch.int32) & 0x7FFF
+        pos_b = torch.clamp(abs_bits[:, None] + deltas[None, :], 0, 0x7BFF)
+        neg_b = pos_b | 0x8000
+        all_b = torch.cat([pos_b, neg_b], dim=1)  # (B, 2*(2*rad+1))
+        all_d = all_b.to(torch.int16).view(torch.float16).to(torch.float32)
+        act_u = torch.abs(all_d) > 1.0e-15
+        sd_u = torch.where(act_u, all_d, torch.ones_like(all_d))[:, :, None]
+        sh_u = (X[:, None, :] / sd_u) + 12582912.0
+        qi_u = ((sh_u.view(torch.int32) & 0x007FFFFF) - 0x00400000 + 8).clamp(0, 15)
+        q_u = torch.where(act_u[:, :, None], qi_u, torch.full_like(qi_u, 8)).to(torch.uint8)
+        e_u = X[:, None, :] - all_d[:, :, None] * (q_u.to(torch.float32) - 8.0)
+        sse_u = torch.sum(W[:, None, :] * e_u * e_u, dim=2)
+        pick_u = torch.argmin(sse_u, dim=1)
+        best_d16 = all_b[rows, pick_u]
+        best_d = all_d[rows, pick_u]
+        best_q = q_u[rows, pick_u].to(torch.int64)
+
+        # 5. Closed-Form 1-D Quadratic Vesica + DC Code Projection within sse_budget:
+        if cfg.use_spectral and (lam_vw > 0.0 or lam_dc > 0.0):
+            d_col = best_d[:, None]
+            e = X - d_col * (best_q.to(torch.float32) - 8.0)
+            sse_live = torch.sum(W * e * e, dim=1)
+            sse_cap = sse_live * (1.0 + float(cfg.sse_budget))
+            v = e[:, :half] + e[:, half:]
+            dc = torch.sum(e, dim=1)
+            denom = (W + lam_vw + lam_dc) * (d_col * d_col)
+            act_b = torch.abs(d_col) > 1.0e-15
+            s_den = torch.where(act_b & (denom > 1.0e-30), denom, torch.ones_like(denom))
+            v_full = torch.cat([v, v], dim=1)
+            grad = d_col * (W * e + lam_vw * v_full + lam_dc * dc[:, None])
+            sh_dq = (grad / s_den) + 12582912.0
+            dq = torch.where(
+                act_b,
+                ((sh_dq.view(torch.int32) & 0x007FFFFF) - 0x00400000).to(torch.int64),
+                torch.zeros_like(best_q),
+            )
+            q_new = torch.clamp(best_q + dq, 0, 15)
+            e_new = X - d_col * (q_new.to(torch.float32) - 8.0)
+            sse_new = torch.sum(W * e_new * e_new, dim=1)
+            v_new = e_new[:, :half] + e_new[:, half:]
+            dc_new = torch.sum(e_new, dim=1)
+            spec_old = lam_vw * torch.sum(v * v, dim=1) + lam_dc * dc * dc
+            spec_new = lam_vw * torch.sum(v_new * v_new, dim=1) + lam_dc * dc_new * dc_new
+            accept = (sse_new <= sse_cap) & ((sse_new + spec_new) < (sse_live + spec_old) - 1.0e-12)
+            best_q = torch.where(accept[:, None], q_new, best_q)
+
+        best_q_u8 = best_q.to(torch.uint8)
+        deq_fin = best_d[:, None] * (best_q_u8.to(torch.float32) - 8.0)
+        raw_sse = float(torch.sum((X - deq_fin) ** 2, dtype=torch.float64).item())
+
+        d16_np = best_d16.to(torch.int16).cpu().numpy().view(np.uint16)
+        q_np = best_q_u8.cpu().numpy()
+        out = np.empty((B, Q4_0_BLOCK_BYTES), dtype=np.uint8)
+        out[:, :2] = d16_np.astype("<u2").view(np.uint8).reshape(B, 2)
+        out[:, 2:] = (q_np[:, :16] & 0x0F) | ((q_np[:, 16:] & 0x0F) << np.uint8(4))
+        return out.tobytes(), raw_sse
+
+
+def quantize_tensor_q4_0(
+    weights: np.ndarray,
+    importance: np.ndarray | None = None,
+    cfg: HPCConfig | None = None,
+    chunk_blocks: int = 131072,
+) -> tuple[bytes, int, float]:
+    """Quantize a tensor to GGML Q4_0 using the full HPC V2 pipeline."""
+    cfg = cfg or HPCConfig()
+    x = np.asarray(weights, dtype=np.float32).ravel()
+    if x.size == 0 or x.size % QK4_0 != 0:
+        raise ValueError("Q4_0 requires a non-empty weight count divisible by 32")
+    n_blocks = x.size // QK4_0
+    X = x.reshape(n_blocks, QK4_0)
+
+    if importance is not None:
+        imp = np.asarray(importance, dtype=np.float32).ravel()
+        if imp.size != x.size:
+            raise ValueError("importance length must match weight count")
+        W_raw = imp.reshape(n_blocks, QK4_0)
+    else:
+        W_raw = None
+
+    use_cuda = cfg.device != "cpu" and cuda_is_available() and (
+        cfg.device.startswith("cuda") or (cfg.device == "auto" and n_blocks >= 64)
+    )
+    dev = torch.device("cuda" if cfg.device == "auto" else cfg.device) if use_cuda else None
+
+    chunks: list[bytes] = []
+    total_sse = 0.0
+    for start in range(0, n_blocks, chunk_blocks):
+        stop = min(start + chunk_blocks, n_blocks)
+        Xb = X[start:stop]
+        sig2 = np.mean(Xb * Xb, axis=1, keepdims=True)
+        if W_raw is None:
+            Wb = np.sqrt(sig2 + Xb * Xb)
+        else:
+            wr = W_raw[start:stop]
+            Wb = np.sqrt(np.maximum(wr, 1.0e-6)) * np.sqrt(sig2 + Xb * Xb) if cfg.temper_imatrix else wr
+        wm = np.mean(Wb, axis=1, keepdims=True)
+        Wb = np.where(wm > 1.0e-30, Wb / np.maximum(wm, 1.0e-30), 1.0).astype(np.float32)
+
+        if use_cuda and dev is not None:
+            raw_b, sse_b = _quantize_q4_0_hpc_cuda(Xb, Wb, dev, cfg)
+        else:
+            raw_b, sse_b = _quantize_q4_0_hpc_cpu(Xb, Wb, cfg)
+        chunks.append(raw_b)
+        total_sse += sse_b
+
+    rmse = float(np.sqrt(total_sse / x.size))
+    return b"".join(chunks), n_blocks, rmse