feat(e-audit-03): checkpoint history UI + compress/strict toggles + mid-run trigger

V7-Q03 (E-AUDIT-03, cppmega-mlx-9avx) — Lane 6 audit gap closure.

Q03.1 ckpt.list_history RPC: scans dir recursively for *.safetensors,
reads metadata via read_ckpt_metadata, sorts by mtime desc, caps at
max_entries. Returns path/mtime/size/arch_hash/opt_kind/global_step/
has_opt_sidecar. Registered in dispatcher.

Q03.2 CheckpointHistoryDropdown component: lazy-fetch on first open,
click row -> onSelect(path) fills train-checkpoint-load-path. Wired
into TopBar (renders only when rpc prop provided).

Q03.3 Compress + strict toggles: train-opt-compress dropdown (none/
weights-int8/opt-fp16/both), train-opt-ckpt-strict checkbox,
train-opt-opt-state-strict checkbox. Threaded through onRunPipeline
to App.tsx then stage_train via stage_options.train.

Q03.4 V7-H05 mid-run checkpoint trigger: new stages.request_checkpoint
+ consume_checkpoint_trigger; train loop drains the queue between
steps and saves weights + opt-state sidecar via the existing
_save_compressed path.

Tests (all green):
- tests/v4/test_ckpt_list_history.py: 7 cases (empty/metadata/
  sidecar/sort/cap/missing/recursive)
- tests/v4/test_ckpt_trigger_mid_run.py: 3 cases (queue roundtrip,
  unknown-token, full end-to-end save during train)
- vbgui/tests/CheckpointHistoryDropdown.test.tsx: 5 cases

Regression: backend pytest checkpoint+train 191/191 PASS (81s);
vbgui vitest 428->440 (+12) PASS.

Closes Lane 6 P1 audit gap from docs/UI-TO-TRAIN-AUDIT-2026-05-23.md.

Author: apstenku123

cppmega_v4/jsonrpc/ckpt_history_method.py

@@ -0,0 +1,126 @@
+"""V7-Q03.1: ckpt.list_history RPC.
+
+Scans a directory recursively for ``*.safetensors`` files, reads
+self-describing metadata (header-only, no full tensor load) and
+returns a sorted-by-mtime descending list capped at 100. Pairs with
+``ckpt.inspect`` for the single-file inspector view and feeds the
+``CheckpointHistoryDropdown`` UI component that lets the operator pick
+a past checkpoint for resume without copy-pasting paths.
+
+Closes Lane 6 audit gap from docs/UI-TO-TRAIN-AUDIT-2026-05-23.md.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from cppmega_v4.jsonrpc.cache import LRUCache
+
+
+class CkptListHistoryParams(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    directory: str = "."
+    # Cap to prevent runaway scans on large workspaces.
+    max_entries: int = 100
+
+
+class CkptHistoryEntry(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    path: str
+    mtime: float
+    size_bytes: int
+    arch_hash: str | None = None
+    opt_kind: str | None = None
+    global_step: int | None = None
+    has_opt_sidecar: bool = False
+
+
+class CkptListHistoryResult(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    directory: str
+    scanned: int = 0
+    entries: list[CkptHistoryEntry] = Field(default_factory=list)
+    error: str | None = None
+
+
+def ckpt_list_history(
+    params: CkptListHistoryParams, *, cache: LRUCache | None = None,
+) -> CkptListHistoryResult:
+    root = Path(params.directory).expanduser()
+    try:
+        root = root.resolve()
+    except Exception as exc:  # noqa: BLE001 -- surface to caller
+        return CkptListHistoryResult(
+            directory=str(params.directory), error=f"resolve failed: {exc}",
+        )
+    if not root.exists():
+        return CkptListHistoryResult(
+            directory=str(root), error="directory does not exist",
+        )
+    if not root.is_dir():
+        return CkptListHistoryResult(
+            directory=str(root), error="path is not a directory",
+        )
+
+    from cppmega_v4.runner.stages import read_ckpt_metadata
+
+    raw_entries: list[CkptHistoryEntry] = []
+    scanned = 0
+    for path in root.rglob("*.safetensors"):
+        # Skip opt-state sidecars — they're surfaced via has_opt_sidecar.
+        name = path.name
+        if name.endswith(".opt.safetensors") or name.endswith(".opt"):
+            continue
+        scanned += 1
+        try:
+            stat = path.stat()
+            meta: dict[str, Any] | None = None
+            try:
+                meta = read_ckpt_metadata(str(path))
+            except Exception:
+                meta = None
+            arch = (meta or {}).get("arch")
+            train = (meta or {}).get("train")
+            opt = (meta or {}).get("opt")
+            arch_hash = (
+                arch.get("config_hash") if isinstance(arch, dict) else None
+            )
+            opt_kind = (
+                opt.get("kind") if isinstance(opt, dict) else None
+            )
+            global_step = (
+                train.get("global_step") if isinstance(train, dict) else None
+            )
+            sidecar = path.with_suffix(path.suffix + ".opt")
+            raw_entries.append(CkptHistoryEntry(
+                path=str(path),
+                mtime=stat.st_mtime,
+                size_bytes=stat.st_size,
+                arch_hash=arch_hash,
+                opt_kind=opt_kind,
+                global_step=global_step,
+                has_opt_sidecar=sidecar.is_file(),
+            ))
+        except Exception:
+            # Unreadable file — skip silently.
+            continue
+
+    raw_entries.sort(key=lambda e: e.mtime, reverse=True)
+    capped = raw_entries[: max(1, int(params.max_entries))]
+    return CkptListHistoryResult(
+        directory=str(root), scanned=scanned, entries=capped,
+    )
+
+
+__all__ = [
+    "CkptListHistoryParams",
+    "CkptHistoryEntry",
+    "CkptListHistoryResult",
+    "ckpt_list_history",
+]

cppmega_v4/jsonrpc/dispatcher.py

@@ -46,6 +46,9 @@
 from cppmega_v4.jsonrpc.ckpt_inspect_method import (
     CkptInspectParams, ckpt_inspect,
 )
+from cppmega_v4.jsonrpc.ckpt_history_method import (
+    CkptListHistoryParams, ckpt_list_history,
+)
 from cppmega_v4.jsonrpc.dtype_cost_method import (
     DtypeCostParams, dtype_cost_estimate,
 )
@@ -176,6 +179,10 @@
         CkptInspectParams,
         lambda p, c: ckpt_inspect(p, cache=c),
     ),
+    "ckpt.list_history": (
+        CkptListHistoryParams,
+        lambda p, c: ckpt_list_history(p, cache=c),
+    ),
     "dtype.cost_estimate": (
         DtypeCostParams,
         lambda p, c: dtype_cost_estimate(p, cache=c),

cppmega_v4/runner/stages.py

@@ -1895,6 +1895,46 @@ def _compiled_step_value_and_grad(_emb, _targets):
                     val_losses.append(vL)
                 except Exception:
                     pass
+            # V7-Q03.4: drain mid-run checkpoint trigger queue. WS
+            # handler can request a save without aborting; we save via
+            # the same _save_compressed path as end-of-run + sidecar.
+            if abort_token is not None:
+                trig_path = consume_checkpoint_trigger(str(abort_token))
+                if trig_path:
+                    try:
+                        from cppmega_v4.runtime.checkpoint_quantize import (
+                            save_state_compressed as _save_mid,
+                        )
+                        mid_flat = dict(nn.utils.tree_flatten(
+                            all_modules.parameters()))
+                        mid_meta = _build_ckpt_metadata(
+                            ctx=ctx, optimizer_kind=optimizer_kind,
+                            n_steps=step + 1, lr=lr,
+                        )
+                        _save_mid(mid_flat, trig_path,
+                                  compress="none",
+                                  metadata=mid_meta,
+                                  role="weights")
+                        # Sidecar opt-state for full resumable bundle.
+                        try:
+                            import safetensors.mlx as _stmlx_mid
+                            mid_opt_flat = dict(nn.utils.tree_flatten(
+                                opt.state))
+                            mid_opt_arrays = {
+                                k: v for k, v in mid_opt_flat.items()
+                                if hasattr(v, "shape")
+                            }
+                            if hasattr(rng_key, "shape"):
+                                mid_opt_arrays["_rng_key"] = rng_key
+                            _stmlx_mid.save_file(
+                                mid_opt_arrays, trig_path + ".opt",
+                                metadata=mid_meta,
+                            )
+                        except Exception:
+                            pass
+                    except Exception:
+                        # Best-effort; don't fail the run on save error.
+                        pass
             if abort_token is not None and abort_token in _ABORT_TOKENS:
                 partial_delta = 0.0
                 if probe_key is not None and probe_before is not None:
@@ -2516,6 +2556,29 @@ def clear_abort(token: str) -> None:
     _ABORT_TOKENS.discard(token)
 
 
+# V7-Q03.4: in-process trigger-checkpoint queue. Caller (WS handler)
+# inserts `{abort_token: save_path}` and the train loop drains it on
+# the next step boundary, saving weights + opt-state sidecar via the
+# existing _save_compressed path. Live mid-run checkpointing without
+# affecting the post-train save_path.
+_TRIGGER_CHECKPOINT_QUEUE: dict[str, str] = {}
+
+
+def request_checkpoint(token: str, save_path: str) -> None:
+    """V7-Q03.4: queue a mid-run checkpoint for the named run.
+
+    The train loop polls _TRIGGER_CHECKPOINT_QUEUE between steps; when
+    it finds an entry keyed by the active abort_token, it saves to
+    save_path (and the sidecar `<save_path>.opt`) then pops the entry.
+    """
+    _TRIGGER_CHECKPOINT_QUEUE[token] = save_path
+
+
+def consume_checkpoint_trigger(token: str) -> str | None:
+    """Pop and return any pending checkpoint save_path for token."""
+    return _TRIGGER_CHECKPOINT_QUEUE.pop(token, None)
+
+
 _REWRITER_FACTORIES: dict[str, Callable[..., Any]] = {}
 
 

tests/v4/test_ckpt_list_history.py

@@ -0,0 +1,114 @@
+"""V7-Q03.1: ckpt.list_history RPC regression.
+
+Pins: scan returns sorted-by-mtime descending list of safetensors files
+with metadata extracted via read_ckpt_metadata. Skips opt-state sidecars.
+Surfaces has_opt_sidecar flag when `<path>.opt` exists alongside.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import time
+
+import mlx.core as mx
+import safetensors.mlx as stmlx
+
+from cppmega_v4.jsonrpc.ckpt_history_method import (
+    CkptListHistoryParams, ckpt_list_history,
+)
+
+
+def _make_ckpt(path: str, *, with_sidecar: bool = False,
+               arch_hash: str = "abc123def456", step: int = 7) -> None:
+    weights = {"layers.0.weight": mx.zeros((4, 4))}
+    meta = {
+        "cppmega_version": "test",
+        "arch": '{"config_hash": "' + arch_hash + '"}',
+        "train": '{"global_step": ' + str(step) + '}',
+        "opt": '{"kind": "adamw", "lr": 0.001}',
+    }
+    stmlx.save_file(weights, path, metadata=meta)
+    if with_sidecar:
+        opt_state = {"opt.0.m": mx.zeros((4, 4))}
+        stmlx.save_file(opt_state, path + ".opt", metadata=meta)
+
+
+def test_list_history_empty_dir() -> None:
+    with tempfile.TemporaryDirectory() as td:
+        res = ckpt_list_history(CkptListHistoryParams(directory=td))
+    assert res.error is None
+    assert res.scanned == 0
+    assert res.entries == []
+
+
+def test_list_history_returns_metadata() -> None:
+    with tempfile.TemporaryDirectory() as td:
+        p = os.path.join(td, "a.safetensors")
+        _make_ckpt(p, arch_hash="aaa111", step=4)
+        res = ckpt_list_history(CkptListHistoryParams(directory=td))
+    assert res.error is None
+    assert res.scanned == 1
+    assert len(res.entries) == 1
+    e = res.entries[0]
+    assert e.path.endswith("a.safetensors")
+    assert e.arch_hash == "aaa111"
+    assert e.opt_kind == "adamw"
+    assert e.global_step == 4
+    assert e.size_bytes > 0
+    assert e.has_opt_sidecar is False
+
+
+def test_list_history_detects_sidecar() -> None:
+    with tempfile.TemporaryDirectory() as td:
+        p = os.path.join(td, "withopt.safetensors")
+        _make_ckpt(p, with_sidecar=True)
+        res = ckpt_list_history(CkptListHistoryParams(directory=td))
+    # Sidecar file is NOT a top-level entry but IS detected on the
+    # parent's has_opt_sidecar flag.
+    assert res.scanned == 1
+    assert len(res.entries) == 1
+    assert res.entries[0].has_opt_sidecar is True
+
+
+def test_list_history_sorted_by_mtime_desc() -> None:
+    with tempfile.TemporaryDirectory() as td:
+        a = os.path.join(td, "older.safetensors")
+        b = os.path.join(td, "newer.safetensors")
+        _make_ckpt(a)
+        time.sleep(0.05)  # mtime granularity
+        _make_ckpt(b)
+        res = ckpt_list_history(CkptListHistoryParams(directory=td))
+    assert [e.path.split("/")[-1] for e in res.entries] == [
+        "newer.safetensors", "older.safetensors",
+    ]
+
+
+def test_list_history_caps_at_max_entries() -> None:
+    with tempfile.TemporaryDirectory() as td:
+        for i in range(5):
+            _make_ckpt(os.path.join(td, f"c{i}.safetensors"))
+        res = ckpt_list_history(
+            CkptListHistoryParams(directory=td, max_entries=3))
+    assert res.scanned == 5
+    assert len(res.entries) == 3
+
+
+def test_list_history_missing_dir() -> None:
+    res = ckpt_list_history(
+        CkptListHistoryParams(directory="/nonexistent/dir/xyzzy"))
+    assert res.error is not None
+    assert "does not exist" in res.error
+
+
+def test_list_history_recursive() -> None:
+    with tempfile.TemporaryDirectory() as td:
+        sub = os.path.join(td, "sub")
+        os.makedirs(sub)
+        _make_ckpt(os.path.join(td, "top.safetensors"))
+        _make_ckpt(os.path.join(sub, "deep.safetensors"))
+        res = ckpt_list_history(CkptListHistoryParams(directory=td))
+    assert res.scanned == 2
+    assert {e.path.split("/")[-1] for e in res.entries} == {
+        "top.safetensors", "deep.safetensors",
+    }