`1fdf194`

Finish CLI M3 coverage pass

Authored by

espadonne 3 weeks ago

SHA: 1fdf1940830d54f7dde07b9eab7c8a5a63598db4
Parents: cda2c03
Tree: 2884406

6 changed files

Status	File	+	-
M	`src/dlm/cli/commands.py`	15	16
M	`tests/integration/audio/test_audio_training.py`	11	4
M	`tests/integration/vl/test_vl_training.py`	11	4
A	`tests/unit/cli/test_export_vl_cache_fallback.py`	74	0
A	`tests/unit/cli/test_misc_commands_coverage.py`	520	0
A	`tests/unit/cli/test_train_prompt_repl_coverage.py`	465	0

src/dlm/cli/commands.pymodified

      store.ensure_layout()
      # VL bases: arch-probe + try single-file GGUF on SUPPORTED (with
 -    # fallback to HF-snapshot on refusal or subprocess failure). We
 -    # still need the resolved plan + cached base dir for the GGUF
 -    # path, so resolve those first, then let the dispatcher decide
 -    # whether to use them.
 +    # fallback to HF-snapshot on refusal or subprocess failure). A
 +    # missing local base snapshot should not hard-fail the whole
 +    # export — the dispatcher can still emit the HF-snapshot path
 +    # without GGUF context.
      if export_dispatch.accepts_images:
 +        gguf_emission_context = None
          try:
              cached_vl = download_spec(spec, local_files_only=True)
          except RuntimeError as exc:
 -            console.print(
 -                f"[red]export:[/red] base model not in local cache "
 -                f"— run `dlm train` first.\n  {exc}"
 -            )
 -            raise typer.Exit(code=1) from exc
 +            _ = exc
 +        else:
 +            gguf_emission_context = {
 +                "plan": plan,
 +                "cached_base_dir": cached_vl.path,
 +                "source_dlm_path": path.resolve(),
 +                "training_sequence_len": parsed.frontmatter.training.sequence_len,
 +                "dlm_version": f"v{parsed.frontmatter.dlm_version}",
 +            }
          try:
              dispatch_result = export_dispatch.dispatch_export(
                  store=store,
                  quant=quant,
                  merged=merged,
                  adapter_mix_raw=adapter_mix,
 -                gguf_emission_context={
 -                    "plan": plan,
 -                    "cached_base_dir": cached_vl.path,
 -                    "source_dlm_path": path.resolve(),
 -                    "training_sequence_len": parsed.frontmatter.training.sequence_len,
 -                    "dlm_version": f"v{parsed.frontmatter.dlm_version}",
 -                },
 +                gguf_emission_context=gguf_emission_context,
+             )
          except ExportError as exc:
              console.print(f"[red]export:[/red] {exc}")

tests/integration/audio/test_audio_training.pymodified

  ) -> None:
      """Full audio cycle: init → ingest wav → train 1 step → verify adapter."""
      import dlm.train as dlm_train
 +    from dlm.base_models import resolve as resolve_base_model
      from dlm.doc.parser import parse_file
 +    from dlm.hardware import doctor
      from dlm.store.manifest import load_manifest
      from dlm.store.paths import for_dlm
      parsed = parse_file(doc)
      store = for_dlm(parsed.frontmatter.dlm_id, home=tmp_home)
 +    spec = resolve_base_model(parsed.frontmatter.base_model, accept_license=True)
 +    plan = doctor(training_config=parsed.frontmatter.training).plan
 +    if plan is None:
 +        pytest.skip("no viable plan on this host — audio body needs a real trainer")
      # Cap steps to 1 so the test completes on commodity hardware.
 -    result = dlm_train.run(
 -        doc,
 +    dlm_train.run(
 +        store,
 +        parsed,
 +        spec,
 +        plan,
          mode="fresh",
          seed=42,
          max_steps=1,
 -        home=tmp_home,
+     )
 -    assert result is not None
      # Adapter committed under v0001/.
      adapter_dir = store.resolve_current_adapter()

tests/integration/vl/test_vl_training.pymodified

      aren't locally cached.
      """
      import dlm.train as dlm_train
 +    from dlm.base_models import resolve as resolve_base_model
      from dlm.doc.parser import parse_file
 +    from dlm.hardware import doctor
      from dlm.store.manifest import load_manifest
      from dlm.store.paths import for_dlm
      parsed = parse_file(doc)
      store = for_dlm(parsed.frontmatter.dlm_id, home=tmp_home)
 +    spec = resolve_base_model(parsed.frontmatter.base_model, accept_license=True)
 +    plan = doctor(training_config=parsed.frontmatter.training).plan
 +    if plan is None:
 +        pytest.skip("no viable plan on this host — VL body needs a real trainer")
      # Cap steps to 1 so the test completes on commodity hardware.
 -    result = dlm_train.run(
 -        doc,
 +    dlm_train.run(
 +        store,
 +        parsed,
 +        spec,
 +        plan,
          mode="fresh",
          seed=42,
          max_steps=1,
 -        home=tmp_home,
+     )
 -    assert result is not None
      # Adapter committed under v0001/.
      adapter_dir = store.resolve_current_adapter()

tests/unit/cli/test_export_vl_cache_fallback.pyadded

 +"""CLI coverage for VL export when GGUF context is unavailable."""
++
 +from __future__ import annotations
++
 +from pathlib import Path
 +from types import SimpleNamespace
 +from typing import Any
++
 +from typer.testing import CliRunner
++
 +from dlm.cli.app import app
++
++
 +def _scaffold_vl_doc(tmp_path: Path) -> Path:
 +    doc = tmp_path / "vl.dlm"
 +    runner = CliRunner()
 +    result = runner.invoke(
 +        app,
 +        [
 +            "--home",
 +            str(tmp_path / "home"),
 +            "init",
 +            str(doc),
 +            "--base",
 +            "paligemma-3b-mix-224",
 +            "--multimodal",
 +            "--i-accept-license",
 +        ],
 +    )
 +    assert result.exit_code == 0, result.output
 +    return doc
++
++
 +class _FakeVlDispatch:
 +    accepts_images = True
 +    accepts_audio = False
++
 +    def __init__(self) -> None:
 +        self.calls: list[dict[str, object]] = []
++
 +    def dispatch_export(self, **kwargs: object) -> object:
 +        self.calls.append(dict(kwargs))
 +        return SimpleNamespace(
 +            export_dir=Path("/tmp/hf-snapshot"),
 +            manifest_path=Path("/tmp/hf-snapshot/snapshot_manifest.json"),
 +            artifacts=[],
 +            banner_lines=["[green]export:[/green] HF snapshot fallback reached"],
 +        )
++
++
 +def test_vl_export_without_cached_base_uses_snapshot_dispatch(
 +    tmp_path: Path,
 +    monkeypatch: Any,
 +) -> None:
 +    doc = _scaffold_vl_doc(tmp_path)
 +    fake_dispatch = _FakeVlDispatch()
++
 +    def _raise_cache_miss(*_: object, **__: object) -> object:
 +        raise RuntimeError("google/paligemma-3b-mix-224 not found in local cache")
++
 +    monkeypatch.setattr("dlm.base_models.download_spec", _raise_cache_miss)
 +    monkeypatch.setattr("dlm.modality.modality_for", lambda _spec: fake_dispatch)
++
 +    runner = CliRunner()
 +    result = runner.invoke(
 +        app,
 +        ["--home", str(tmp_path / "home"), "export", str(doc)],
 +    )
++
 +    assert result.exit_code == 0, result.output
 +    assert len(fake_dispatch.calls) == 1
 +    assert fake_dispatch.calls[0]["gguf_emission_context"] is None
 +    assert "HF snapshot fallback reached" in result.output
 +    assert "base model not in local cache" not in result.output

tests/unit/cli/test_misc_commands_coverage.pyadded

 +"""Coverage-oriented command tests for the remaining CLI surface."""
++
 +from __future__ import annotations
++
 +import json
 +from pathlib import Path
 +from types import SimpleNamespace
 +from typing import Any
++
 +import pytest
 +from click.exceptions import Exit as ClickExit
 +from typer.testing import CliRunner
++
 +from dlm.cli import commands
 +from dlm.cli.app import app
 +from dlm.share.signing import VerifyStatus
 +from dlm.store.paths import for_dlm
++
++
 +def _write_minimal_dlm(
 +    path: Path,
 +    *,
 +    dlm_id: str = "01KPQ9M3" + "0" * 18,
 +    base_model: str = "smollm2-135m",
 +) -> None:
 +    path.write_text(
 +        f"---\ndlm_id: {dlm_id}\ndlm_version: 12\nbase_model: {base_model}\n---\nbody\n",
 +        encoding="utf-8",
 +    )
++
++
 +def _joined_output(result: object) -> str:
 +    text = getattr(result, "output", "") + getattr(result, "stderr", "")
 +    return " ".join(text.split())
++
++
 +class TestMetricsAndDoctor:
 +    def test_metrics_run_id_json_and_csv(self, tmp_path: Path, monkeypatch: Any) -> None:
 +        doc = tmp_path / "doc.dlm"
 +        _write_minimal_dlm(doc)
 +        monkeypatch.setenv("DLM_HOME", str(tmp_path / "home"))
++
 +        run = SimpleNamespace(
 +            run_id=7,
 +            phase="sft",
 +            seed=42,
 +            status="ok",
 +            started_at="2026-04-21T10:00:00Z",
 +            ended_at="2026-04-21T10:01:00Z",
 +        )
 +        steps = [SimpleNamespace(step=1, loss=0.5, lr=1e-4, grad_norm=0.9)]
 +        evals = [SimpleNamespace(step=1, val_loss=0.4, perplexity=1.5)]
++
 +        monkeypatch.setattr("dlm.metrics.queries.recent_runs", lambda *args, **kwargs: [run])
 +        monkeypatch.setattr("dlm.metrics.queries.steps_for_run", lambda *args, **kwargs: steps)
 +        monkeypatch.setattr("dlm.metrics.queries.evals_for_run", lambda *args, **kwargs: evals)
 +        monkeypatch.setattr(
 +            "dlm.metrics.queries.runs_to_dict",
 +            lambda runs: [
 +                {
 +                    "run_id": r.run_id,
 +                    "phase": r.phase,
 +                    "seed": r.seed,
 +                    "status": r.status,
 +                }
 +                for r in runs
 +            ],
 +        )
 +        monkeypatch.setattr(
 +            "dlm.metrics.queries.steps_to_dict",
 +            lambda rows: [{"step": r.step, "loss": r.loss} for r in rows],
 +        )
 +        monkeypatch.setattr(
 +            "dlm.metrics.queries.evals_to_dict",
 +            lambda rows: [{"step": r.step, "val_loss": r.val_loss} for r in rows],
 +        )
++
 +        import sys
 +        from io import StringIO
++
 +        old_stdout = sys.stdout
 +        try:
 +            json_buf = StringIO()
 +            sys.stdout = json_buf
 +            commands.metrics_cmd(doc, json_out=True, run_id=7)
 +        finally:
 +            sys.stdout = old_stdout
 +        payload = json.loads(json_buf.getvalue())
 +        assert payload["run"]["run_id"] == 7
 +        assert payload["steps"][0]["step"] == 1
 +        assert payload["evals"][0]["val_loss"] == 0.4
++
 +        old_stdout = sys.stdout
 +        try:
 +            csv_buf = StringIO()
 +            sys.stdout = csv_buf
 +            commands.metrics_cmd(doc, csv_out=True, run_id=7)
 +        finally:
 +            sys.stdout = old_stdout
 +        csv_text = csv_buf.getvalue()
 +        assert "step,loss,lr,grad_norm,val_loss" in csv_text
 +        assert "1,0.5,0.0001,0.9,0.4" in csv_text
++
 +    def test_metrics_since_parse_and_watch(
 +        self, tmp_path: Path, monkeypatch: Any, capsys: Any
 +    ) -> None:
 +        doc = tmp_path / "doc.dlm"
 +        home = tmp_path / "home"
 +        _write_minimal_dlm(doc)
 +        monkeypatch.setenv("DLM_HOME", str(home))
++
 +        with pytest.raises(ClickExit) as excinfo:
 +            commands.metrics_cmd(doc, since="bogus")
 +        assert excinfo.type is ClickExit
 +        err = capsys.readouterr().err
 +        assert "not an integer+unit" in err
++
 +        monkeypatch.setattr("dlm.metrics.queries.latest_run_id", lambda *_: 7)
 +        monkeypatch.setattr(
 +            "dlm.metrics.queries.steps_for_run",
 +            lambda *args, **kwargs: [SimpleNamespace(step=3, loss=0.25, lr=1e-4, grad_norm=0.8)],
 +        )
 +        monkeypatch.setattr(
 +            "dlm.metrics.queries.evals_for_run",
 +            lambda *args, **kwargs: [SimpleNamespace(step=3, val_loss=0.2, perplexity=1.2)],
 +        )
++
 +        def _interrupt(*_: object) -> None:
 +            raise KeyboardInterrupt
++
 +        monkeypatch.setattr("time.sleep", _interrupt)
 +        commands.metrics_watch_cmd(doc, poll_seconds=0.0)
 +        out = capsys.readouterr().out
 +        assert "following run_id=7" in out
 +        assert "eval @ step 3" in out
 +        assert "bye" in out
++
 +    def test_doctor_json_and_text(self, monkeypatch: Any, capsys: Any) -> None:
 +        fake = SimpleNamespace(to_dict=lambda: {"plan": "ok"})
 +        monkeypatch.setattr("dlm.hardware.doctor", lambda: fake)
 +        monkeypatch.setattr("dlm.hardware.render_text", lambda result: f"doctor text: {result!r}")
++
 +        commands.doctor_cmd(json_out=True)
 +        json_out = capsys.readouterr().out
 +        assert '"plan": "ok"' in json_out
++
 +        commands.doctor_cmd(json_out=False)
 +        text_out = capsys.readouterr().out
 +        assert "doctor text" in text_out
++
++
 +class TestPackUnpackVerify:
 +    def test_pack_and_unpack_success(self, tmp_path: Path, monkeypatch: Any) -> None:
 +        runner = CliRunner()
 +        pack_out = tmp_path / "bundle.dlm.pack"
 +        monkeypatch.setattr(
 +            "dlm.pack.packer.pack",
 +            lambda *args, **kwargs: SimpleNamespace(
 +                path=pack_out,
 +                bytes_written=5 * 1024 * 1024,
 +                content_type="application/dlm-pack",
 +            ),
 +        )
 +        pack_result = runner.invoke(
 +            app, ["pack", str(tmp_path / "doc.dlm"), "--out", str(pack_out)]
 +        )
 +        assert pack_result.exit_code == 0, pack_result.output
 +        assert "packed:" in pack_result.output
 +        assert "application/dlm-pack" in pack_result.output
++
 +        monkeypatch.setattr(
 +            "dlm.pack.unpacker.unpack",
 +            lambda *args, **kwargs: SimpleNamespace(
 +                dlm_path=tmp_path / "restored.dlm",
 +                store_path=tmp_path / "home" / "store" / "01XYZ",
 +                dlm_id="01XYZ",
 +                header=SimpleNamespace(pack_format_version=2),
 +                applied_migrations=[2, 3],
 +            ),
 +        )
 +        unpack_result = runner.invoke(
 +            app,
 +            ["unpack", str(pack_out), "--out", str(tmp_path / "restore")],
 +        )
 +        assert unpack_result.exit_code == 0, unpack_result.output
 +        assert "unpacked:" in unpack_result.output
 +        assert "migrated: v2" in unpack_result.output
++
 +    def test_verify_success_and_unsigned(self, tmp_path: Path, monkeypatch: Any) -> None:
 +        runner = CliRunner()
 +        pack_path = tmp_path / "bundle.dlm.pack"
 +        pack_path.write_bytes(b"pack")
++
 +        provenance = SimpleNamespace(
 +            adapter_sha256="a" * 64,
 +            base_revision="main",
 +            corpus_root_sha256="b" * 64,
 +            signed_at="2026-04-21T12:00:00Z",
 +        )
 +        monkeypatch.setattr(
 +            "dlm.pack.unpacker.read_pack_member_bytes", lambda *args, **kwargs: b"{}"
 +        )
 +        monkeypatch.setattr("dlm.share.provenance.load_provenance_json", lambda *_: provenance)
 +        monkeypatch.setattr(
 +            "dlm.share.provenance.verify_provenance",
 +            lambda *args, **kwargs: SimpleNamespace(
 +                signer_fingerprint="FINGERPRINT",
 +                trusted_key_path=tmp_path / "trusted.pub",
 +                tofu_recorded=True,
 +            ),
 +        )
++
 +        result = runner.invoke(app, ["verify", str(pack_path), "--trust-on-first-use"])
 +        assert result.exit_code == 0, result.output
 +        assert "verified:" in result.output
 +        assert "recorded new trust entry" in result.output
++
 +        monkeypatch.setattr(
 +            "dlm.pack.unpacker.read_pack_member_bytes",
 +            lambda *args, **kwargs: None,
 +        )
 +        unsigned = runner.invoke(app, ["verify", str(pack_path)])
 +        assert unsigned.exit_code == 1, unsigned.output
 +        assert "is unsigned" in unsigned.output
++
++
 +class TestMigrateTemplatesShareAndServe:
 +    def test_migrate_templates_push_pull_and_serve(self, tmp_path: Path, monkeypatch: Any) -> None:
 +        runner = CliRunner()
 +        doc = tmp_path / "doc.dlm"
 +        home = tmp_path / "home"
 +        _write_minimal_dlm(doc)
++
 +        monkeypatch.setattr(
 +            "dlm.doc.migrate.migrate_file",
 +            lambda *args, **kwargs: SimpleNamespace(
 +                applied=[11],
 +                target_version=12,
 +                backup_path=tmp_path / "doc.dlm.bak",
 +            ),
 +        )
 +        migrate_result = runner.invoke(app, ["migrate", str(doc)])
 +        assert migrate_result.exit_code == 0, migrate_result.output
 +        assert "migrated:" in migrate_result.output
++
 +        template = SimpleNamespace(
 +            name="starter",
 +            meta=SimpleNamespace(
 +                title="Starter",
 +                domain_tags=("docs",),
 +                recommended_base="smollm2-135m",
 +                expected_steps=25,
 +                expected_duration={"minutes": 5},
 +                summary="Starter template",
 +                sample_prompts=("hello",),
 +            ),
 +        )
 +        monkeypatch.setattr("dlm.templates.list_bundled", lambda: [template])
 +        templates_result = runner.invoke(app, ["templates", "list", "--json"])
 +        assert templates_result.exit_code == 0, templates_result.output
 +        assert '"name":"starter"' in templates_result.output.replace(" ", "")
++
 +        templates_text = runner.invoke(app, ["templates", "list"])
 +        assert templates_text.exit_code == 0, templates_text.output
 +        assert "Starter" in templates_text.output
 +        assert "smollm2-135m" in templates_text.output
++
 +        from dlm.share.sinks import SinkKind
++
 +        monkeypatch.setattr(
 +            "dlm.share.push",
 +            lambda *args, **kwargs: SimpleNamespace(
 +                destination="hf:org/repo",
 +                bytes_sent=3 * 1024 * 1024,
 +                sink_kind=SinkKind.HF,
 +                detail="done",
 +            ),
 +        )
 +        push_result = runner.invoke(
 +            app,
 +            ["push", str(doc), "--to", "hf:org/repo"],
 +        )
 +        assert push_result.exit_code == 0, push_result.output
 +        assert "pushed:" in push_result.output
 +        assert "dlm pull hf:org/repo" in push_result.output
++
 +        monkeypatch.setattr(
 +            "dlm.share.pull",
 +            lambda *args, **kwargs: SimpleNamespace(
 +                source="hf:org/repo",
 +                dlm_path=tmp_path / "restored.dlm",
 +                bytes_received=2 * 1024 * 1024,
 +                verification=SimpleNamespace(
 +                    status=VerifyStatus.VERIFIED,
 +                    key_path=tmp_path / "trusted.pub",
 +                    detail="",
 +                ),
 +            ),
 +        )
 +        pull_result = runner.invoke(app, ["pull", "hf:org/repo"])
 +        assert pull_result.exit_code == 0, pull_result.output
 +        assert "pulled:" in pull_result.output
 +        assert "verified:" in pull_result.output
++
 +        store = for_dlm("01KPQ9M3" + "0" * 18, home=home)
 +        store.ensure_layout()
 +        store.manifest.write_text("{}", encoding="utf-8")
++
 +        def _fake_pack(path: Path, *, out: Path) -> None:
 +            out.write_bytes(b"pack")
++
 +        handle = SimpleNamespace(
 +            bind_host="127.0.0.1",
 +            port=7337,
 +            peer_url="peer://127.0.0.1:7337/01KPQ9M3",
 +            wait_shutdown=lambda: None,
 +        )
 +        monkeypatch.setattr("dlm.pack.packer.pack", _fake_pack)
 +        monkeypatch.setattr("dlm.share.serve", lambda *args, **kwargs: handle)
 +        serve_result = runner.invoke(
 +            app,
 +            ["--home", str(home), "serve", str(doc)],
 +        )
 +        assert serve_result.exit_code == 0, serve_result.output
 +        assert "serving:" in serve_result.output
 +        assert "peer URL:" in serve_result.output
++
 +    def test_templates_refresh_empty_gallery_and_pull_statuses(
 +        self,
 +        tmp_path: Path,
 +        monkeypatch: Any,
 +    ) -> None:
 +        runner = CliRunner()
++
 +        from dlm.templates.fetcher import RemoteFetchUnavailable
++
 +        monkeypatch.setattr(
 +            "dlm.templates.fetcher.fetch_all",
 +            lambda *args, **kwargs: (_ for _ in ()).throw(RemoteFetchUnavailable("offline")),
 +        )
 +        monkeypatch.setattr("dlm.templates.fetcher.cache_dir", lambda: tmp_path / "cache")
 +        monkeypatch.setattr("dlm.templates.list_bundled", lambda: [])
 +        empty = runner.invoke(app, ["templates", "list", "--refresh"])
 +        assert empty.exit_code == 1, empty.output
 +        joined = _joined_output(empty)
 +        assert "Falling back to the bundled gallery" in joined
 +        assert "no bundled templates found" in joined
++
 +        monkeypatch.setattr(
 +            "dlm.share.pull",
 +            lambda *args, **kwargs: SimpleNamespace(
 +                source="hf:org/repo",
 +                dlm_path=tmp_path / "restored.dlm",
 +                bytes_received=1 * 1024 * 1024,
 +                verification=SimpleNamespace(
 +                    status=VerifyStatus.UNVERIFIED,
 +                    key_path=None,
 +                    detail="missing key",
 +                ),
 +            ),
 +        )
 +        unverified = runner.invoke(app, ["pull", "hf:org/repo"])
 +        assert unverified.exit_code == 0, unverified.output
 +        assert "unverified:" in unverified.output
++
 +        monkeypatch.setattr(
 +            "dlm.share.pull",
 +            lambda *args, **kwargs: SimpleNamespace(
 +                source="hf:org/repo",
 +                dlm_path=tmp_path / "restored.dlm",
 +                bytes_received=1 * 1024 * 1024,
 +                verification=SimpleNamespace(
 +                    status=VerifyStatus.UNSIGNED,
 +                    key_path=None,
 +                    detail="",
 +                ),
 +            ),
 +        )
 +        unsigned = runner.invoke(app, ["pull", "hf:org/repo"])
 +        assert unsigned.exit_code == 0, unsigned.output
 +        assert "unsigned" in unsigned.output
++
++
 +class TestExportAndCacheCoverage:
 +    def test_export_standard_path_success_and_errors(
 +        self, tmp_path: Path, monkeypatch: Any
 +    ) -> None:
 +        doc = tmp_path / "doc.dlm"
 +        runner = CliRunner()
 +        result = runner.invoke(
 +            app,
 +            [
 +                "--home",
 +                str(tmp_path / "home"),
 +                "init",
 +                str(doc),
 +                "--base",
 +                "smollm2-135m",
 +            ],
 +        )
 +        assert result.exit_code == 0, result.output
++
 +        cached = SimpleNamespace(path=tmp_path / "base")
 +        monkeypatch.setattr("dlm.base_models.download_spec", lambda *args, **kwargs: cached)
 +        monkeypatch.setattr(
 +            "dlm.export.run_export",
 +            lambda *args, **kwargs: SimpleNamespace(
 +                cached=True,
 +                export_dir=tmp_path / "exports" / "Q4_K_M",
 +                artifacts=[SimpleNamespace(name="base.gguf"), SimpleNamespace(name="adapter.gguf")],
 +                ollama_name="my-model",
 +                ollama_version=1,
 +                smoke_output_first_line="hello",
 +            ),
 +        )
 +        export_ok = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "export", str(doc), "--skip-ollama"],
 +        )
 +        assert export_ok.exit_code == 0, export_ok.output
 +        assert "exported:" in export_ok.output
 +        assert "ollama:" in export_ok.output
 +        assert "smoke:" in export_ok.output
++
 +        monkeypatch.setattr(
 +            "dlm.base_models.download_spec",
 +            lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError("offline cache miss")),
 +        )
 +        cache_miss = runner.invoke(app, ["--home", str(tmp_path / "home"), "export", str(doc)])
 +        assert cache_miss.exit_code == 1, cache_miss.output
 +        assert "base model not in local cache" in cache_miss.output
++
 +        from dlm.export.ollama.errors import OllamaBinaryNotFoundError
++
 +        monkeypatch.setattr("dlm.base_models.download_spec", lambda *args, **kwargs: cached)
 +        monkeypatch.setattr(
 +            "dlm.export.run_export",
 +            lambda *args, **kwargs: (_ for _ in ()).throw(OllamaBinaryNotFoundError("missing")),
 +        )
 +        no_ollama = runner.invoke(app, ["--home", str(tmp_path / "home"), "export", str(doc)])
 +        assert no_ollama.exit_code == 1, no_ollama.output
 +        assert "install from https://ollama.com/download" in no_ollama.output
++
 +    def test_cache_show_prune_and_clear(self, tmp_path: Path, monkeypatch: Any) -> None:
 +        doc = tmp_path / "doc.dlm"
 +        _write_minimal_dlm(doc)
++
 +        class _FakeCache:
 +            def __init__(self, entry_count: int, total_bytes: int) -> None:
 +                self.entry_count = entry_count
 +                self.total_bytes = total_bytes
++
 +            def prune(self, *, older_than_seconds: float) -> int:
 +                self.pruned_seconds = older_than_seconds
 +                return 2
++
 +            def clear(self) -> int:
 +                return 5
++
 +            def save_manifest(self) -> None:
 +                return None
++
 +        fake_cache = _FakeCache(entry_count=5, total_bytes=2048)
 +        monkeypatch.setattr("dlm.directives.cache.TokenizedCache.open", lambda *_: fake_cache)
 +        monkeypatch.setattr(
 +            "dlm.metrics.queries.latest_tokenization",
 +            lambda *_: SimpleNamespace(run_id=9, hit_rate=0.75, cache_hits=3, cache_misses=1),
 +        )
++
 +        runner = CliRunner()
 +        show_json = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "cache", "show", str(doc), "--json"],
 +        )
 +        assert show_json.exit_code == 0, show_json.output
 +        assert '"entry_count": 5' in show_json.output
++
 +        monkeypatch.setattr("dlm.metrics.queries.latest_tokenization", lambda *_: None)
 +        show_text = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "cache", "show", str(doc)],
 +        )
 +        assert show_text.exit_code == 0, show_text.output
 +        assert "Cache for" in show_text.output
 +        assert "no tokenization runs yet" in show_text.output
++
 +        prune_bad = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "cache", "prune", str(doc), "--older-than", "bad"],
 +        )
 +        assert prune_bad.exit_code == 2, prune_bad.output
 +        assert "invalid --older-than" in prune_bad.output
++
 +        prune_ok = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "cache", "prune", str(doc)],
 +        )
 +        assert prune_ok.exit_code == 0, prune_ok.output
 +        assert "older than 90d" in prune_ok.output
++
 +        monkeypatch.setattr("typer.confirm", lambda prompt: False)
 +        clear_cancel = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "cache", "clear", str(doc)],
 +        )
 +        assert clear_cancel.exit_code == 0, clear_cancel.output
 +        assert "clear cancelled" in clear_cancel.output
++
 +        clear_force = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "cache", "clear", str(doc), "--force"],
 +        )
 +        assert clear_force.exit_code == 0, clear_force.output
 +        assert "cleared 5" in clear_force.output
++
 +    def test_parse_duration_helper(self) -> None:
 +        assert commands._parse_duration("30m") == 1800.0
 +        assert commands._parse_duration("2h") == 7200.0
 +        assert commands._parse_duration("1d") == 86400.0
 +        assert commands._parse_duration("7x") is None

tests/unit/cli/test_train_prompt_repl_coverage.pyadded

 +"""Coverage-oriented tests for train/prompt/repl command bodies."""
++
 +from __future__ import annotations
++
 +from pathlib import Path
 +from types import SimpleNamespace
 +from typing import Any
++
 +from typer.testing import CliRunner
++
 +from dlm.cli.app import app
++
++
 +def _init_doc(tmp_path: Path, *, base: str = "smollm2-135m") -> Path:
 +    doc = tmp_path / "doc.dlm"
 +    runner = CliRunner()
 +    result = runner.invoke(
 +        app,
 +        [
 +            "--home",
 +            str(tmp_path / "home"),
 +            "init",
 +            str(doc),
 +            "--base",
 +            base,
 +        ],
 +    )
 +    assert result.exit_code == 0, result.output
 +    return doc
++
++
 +def _fake_doctor_result() -> object:
 +    return SimpleNamespace(plan=object(), capabilities=object())
++
++
 +class TestTrainCommandCoverage:
 +    def test_train_success_prints_phase_summary(self, tmp_path: Path, monkeypatch: Any) -> None:
 +        doc = _init_doc(tmp_path)
 +        runner = CliRunner()
++
 +        fake_result = SimpleNamespace(
 +            adapter_version=1,
 +            steps=3,
 +            seed=42,
 +            determinism=SimpleNamespace(class_="strict"),
 +            adapter_path=tmp_path / "adapter",
 +            log_path=tmp_path / "train.jsonl",
 +            final_train_loss=0.125,
 +        )
 +        fake_phase = SimpleNamespace(phase="sft", result=fake_result)
++
 +        monkeypatch.setattr("dlm.hardware.doctor", lambda **kwargs: _fake_doctor_result())
 +        monkeypatch.setattr("dlm.train.distributed.detect_world_size", lambda: 1)
 +        monkeypatch.setattr(
 +            "dlm.train.preference.phase_orchestrator.run_phases",
 +            lambda *args, **kwargs: [fake_phase],
 +        )
++
 +        result = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "train", str(doc), "--max-steps", "3"],
 +        )
 +        assert result.exit_code == 0, result.output
 +        assert "sft:" in result.output
 +        assert "adapter:" in result.output
 +        assert "0.125" in result.output
++
 +    def test_train_watch_with_rpc_starts_server(self, tmp_path: Path, monkeypatch: Any) -> None:
 +        doc = _init_doc(tmp_path)
 +        runner = CliRunner()
++
 +        fake_result = SimpleNamespace(
 +            adapter_version=1,
 +            steps=1,
 +            seed=7,
 +            determinism=SimpleNamespace(class_="strict"),
 +            adapter_path=tmp_path / "adapter",
 +            log_path=tmp_path / "train.jsonl",
 +            final_train_loss=None,
 +        )
 +        fake_phase = SimpleNamespace(phase="sft", result=fake_result)
++
 +        class _FakeQueue:
 +            capacity = 123
++
 +            def drain(self) -> list[object]:
 +                return []
++
 +        class _FakeServer:
 +            def __init__(self, *, host: str, port: int, token: str, queue: object) -> None:
 +                self.address = (host, port)
++
 +            def start(self) -> None:
 +                return None
++
 +            def stop(self) -> None:
 +                return None
++
 +        monkeypatch.setenv("DLM_PROBE_TOKEN", "secret")
 +        monkeypatch.setattr("dlm.hardware.doctor", lambda **kwargs: _fake_doctor_result())
 +        monkeypatch.setattr("dlm.train.distributed.detect_world_size", lambda: 1)
 +        monkeypatch.setattr(
 +            "dlm.train.preference.phase_orchestrator.run_phases",
 +            lambda *args, **kwargs: [fake_phase],
 +        )
 +        monkeypatch.setattr("dlm.train.inject.InjectedProbeQueue", _FakeQueue)
 +        monkeypatch.setattr("dlm.train.rpc.ProbeRpcServer", _FakeServer)
 +        monkeypatch.setattr("dlm.watch.loop.run_watch", lambda *args, **kwargs: 0)
++
 +        result = runner.invoke(
 +            app,
 +            [
 +                "--home",
 +                str(tmp_path / "home"),
 +                "train",
 +                str(doc),
 +                "--watch",
 +                "--listen-rpc",
 +                "127.0.0.1:7777",
 +            ],
 +        )
 +        assert result.exit_code == 0, result.output
 +        assert "rpc:" in result.output
 +        assert "watch:" in result.output
++
 +    def test_train_noop_watch_repl_and_bounded_rpc_refusals(
 +        self,
 +        tmp_path: Path,
 +        monkeypatch: Any,
 +    ) -> None:
 +        doc = _init_doc(tmp_path)
 +        runner = CliRunner()
++
 +        monkeypatch.setattr("dlm.hardware.doctor", lambda **kwargs: _fake_doctor_result())
 +        monkeypatch.setattr("dlm.train.distributed.detect_world_size", lambda: 1)
 +        monkeypatch.setattr(
 +            "dlm.train.preference.phase_orchestrator.run_phases",
 +            lambda *args, **kwargs: [],
 +        )
++
 +        no_op = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "train", str(doc)],
 +        )
 +        assert no_op.exit_code == 0, no_op.output
 +        assert "nothing to train" in no_op.output
++
 +        fake_result = SimpleNamespace(
 +            adapter_version=1,
 +            steps=1,
 +            seed=42,
 +            determinism=SimpleNamespace(class_="strict"),
 +            adapter_path=tmp_path / "adapter",
 +            log_path=tmp_path / "train.jsonl",
 +            final_train_loss=None,
 +        )
 +        fake_phase = SimpleNamespace(phase="sft", result=fake_result)
 +        monkeypatch.setattr(
 +            "dlm.train.preference.phase_orchestrator.run_phases",
 +            lambda *args, **kwargs: [fake_phase],
 +        )
++
 +        watch_repl = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "train", str(doc), "--watch", "--repl"],
 +        )
 +        assert watch_repl.exit_code == 2, watch_repl.output
 +        assert "not yet implemented" in watch_repl.output
++
 +        monkeypatch.setenv("DLM_PROBE_TOKEN", "secret")
 +        bounded_rpc = runner.invoke(
 +            app,
 +            [
 +                "--home",
 +                str(tmp_path / "home"),
 +                "train",
 +                str(doc),
 +                "--listen-rpc",
 +                "127.0.0.1:7777",
 +                "--max-cycles",
 +                "1",
 +            ],
 +        )
 +        assert bounded_rpc.exit_code == 2, bounded_rpc.output
 +        assert "--watch for now" in bounded_rpc.output
++
 +    def test_multi_gpu_helper_and_strip(self, monkeypatch: Any) -> None:
 +        from rich.console import Console
++
 +        from dlm.cli.commands import _maybe_dispatch_multi_gpu, _strip_gpus_from_argv
 +        from dlm.train.distributed import UnsupportedGpuSpecError
++
 +        class _GpuSpec:
 +            def __init__(self, device_ids: tuple[int, ...]) -> None:
 +                self._device_ids = device_ids
++
 +            def resolve(self, device_count: int) -> tuple[int, ...]:
 +                return self._device_ids
++
 +        console = Console(stderr=True)
 +        monkeypatch.setattr(
 +            "dlm.train.distributed.parse_gpus",
 +            lambda raw: (_ for _ in ()).throw(UnsupportedGpuSpecError("bad gpus")),
 +        )
 +        assert _maybe_dispatch_multi_gpu("bogus", ["dlm", "train"], console) == 2
++
 +        monkeypatch.setattr("dlm.train.distributed.parse_gpus", lambda raw: _GpuSpec((0,)))
 +        import torch
++
 +        monkeypatch.setattr(torch.cuda, "device_count", lambda: 2)
 +        assert _maybe_dispatch_multi_gpu("1", ["dlm", "train"], console) is None
++
 +        launched: dict[str, object] = {}
 +        monkeypatch.setattr("dlm.train.distributed.parse_gpus", lambda raw: _GpuSpec((1, 3)))
 +        monkeypatch.setattr(
 +            "dlm.train.distributed.launch_multi_gpu",
 +            lambda device_ids, cli_args, mixed_precision="bf16": (
 +                launched.update(
 +                    {
 +                        "device_ids": device_ids,
 +                        "cli_args": cli_args,
 +                        "mixed_precision": mixed_precision,
 +                    }
 +                )
 +                or 17
 +            ),
 +        )
 +        exit_code = _maybe_dispatch_multi_gpu(
 +            "1,3",
 +            ["dlm", "train", "doc.dlm", "--gpus", "1,3"],
 +            console,
 +        )
 +        assert exit_code == 17
 +        assert launched["device_ids"] == (1, 3)
 +        assert launched["cli_args"] == ["train", "doc.dlm"]
 +        assert _strip_gpus_from_argv(["dlm", "train", "--gpus=0,1", "doc.dlm"]) == [
 +            "train",
 +            "doc.dlm",
 +        ]
++
 +    def test_train_error_mappings(self, tmp_path: Path, monkeypatch: Any) -> None:
 +        doc = _init_doc(tmp_path)
 +        runner = CliRunner()
++
 +        from dlm.lock.errors import LockValidationError
 +        from dlm.train.errors import DiskSpaceError, OOMError, ResumeIntegrityError, TrainingError
 +        from dlm.train.preference.errors import (
 +            DpoPhaseError,
 +            NoPreferenceContentError,
 +            PriorAdapterRequiredError,
 +        )
++
 +        monkeypatch.setattr("dlm.hardware.doctor", lambda **kwargs: _fake_doctor_result())
 +        monkeypatch.setattr("dlm.train.distributed.detect_world_size", lambda: 1)
++
 +        cases = [
 +            (
 +                LockValidationError(path=tmp_path / "dlm.lock", reasons=["torch drift"]),
 +                "Re-run with",
 +            ),
 +            (DiskSpaceError(required_bytes=2_000_000_000, free_bytes=1_000_000_000), "disk:"),
 +            (ResumeIntegrityError("resume mismatch"), "resume:"),
 +            (NoPreferenceContentError("no preferences"), "dpo:"),
 +            (PriorAdapterRequiredError("need prior adapter"), "dpo:"),
 +            (DpoPhaseError("dpo failed"), "dpo:"),
 +            (TrainingError("trainer failed"), "training:"),
 +        ]
 +        for error, needle in cases:
 +            monkeypatch.setattr(
 +                "dlm.train.preference.phase_orchestrator.run_phases",
 +                lambda *args, _error=error, **kwargs: (_ for _ in ()).throw(_error),
 +            )
 +            result = runner.invoke(
 +                app,
 +                ["--home", str(tmp_path / "home"), "train", str(doc)],
 +            )
 +            assert result.exit_code == 1, result.output
 +            assert needle in result.output
++
 +        monkeypatch.setattr(
 +            "dlm.train.preference.phase_orchestrator.run_phases",
 +            lambda *args, **kwargs: (_ for _ in ()).throw(
 +                OOMError(
 +                    step=5,
 +                    peak_bytes=2_000,
 +                    free_at_start_bytes=4_000,
 +                    current_grad_accum=1,
 +                    recommended_grad_accum=4,
 +                )
 +            ),
 +        )
 +        monkeypatch.setattr("dlm.train.format_oom_message", lambda **kwargs: "OOM advice")
 +        oom = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "train", str(doc)],
 +        )
 +        assert oom.exit_code == 1, oom.output
 +        assert "OOM advice" in oom.output
++
++
 +class TestPromptAndReplCoverage:
 +    def test_prompt_text_backend_reads_stdin_and_generates(
 +        self, tmp_path: Path, monkeypatch: Any
 +    ) -> None:
 +        doc = _init_doc(tmp_path)
 +        runner = CliRunner()
++
 +        class _FakeBackend:
 +            def load(self, spec: object, store: object, adapter_name: str | None = None) -> None:
 +                return None
++
 +            def generate(self, query: str, **kwargs: object) -> str:
 +                return f"reply:{query}"
++
 +        monkeypatch.setattr("dlm.hardware.doctor", lambda: SimpleNamespace(capabilities=object()))
 +        monkeypatch.setattr(
 +            "dlm.inference.backends.select_backend", lambda *args, **kwargs: "pytorch"
 +        )
 +        monkeypatch.setattr(
 +            "dlm.inference.backends.build_backend", lambda *args, **kwargs: _FakeBackend()
 +        )
++
 +        result = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "prompt", str(doc)],
 +            input="hello from stdin\n",
 +        )
 +        assert result.exit_code == 0, result.output
 +        assert "reply:hello from stdin" in result.output
++
 +    def test_repl_success_and_adapter_validation(self, tmp_path: Path, monkeypatch: Any) -> None:
 +        doc = _init_doc(tmp_path)
 +        runner = CliRunner()
++
 +        adapter_bad = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "repl", str(doc), "--adapter", "knowledge"],
 +        )
 +        assert adapter_bad.exit_code == 2, adapter_bad.output
 +        assert "only valid on multi-adapter" in adapter_bad.output
++
 +        class _FakeBackend:
 +            def __init__(self) -> None:
 +                self._loaded = SimpleNamespace(tokenizer="tok")
++
 +            def load(self, spec: object, store: object, adapter_name: str | None = None) -> None:
 +                return None
++
 +        monkeypatch.setattr("dlm.hardware.doctor", lambda: SimpleNamespace(capabilities=object()))
 +        monkeypatch.setattr(
 +            "dlm.inference.backends.select_backend", lambda *args, **kwargs: "pytorch"
 +        )
 +        monkeypatch.setattr(
 +            "dlm.inference.backends.build_backend", lambda *args, **kwargs: _FakeBackend()
 +        )
 +        monkeypatch.setattr("dlm.repl.app.run_repl", lambda session, console: 5)
++
 +        repl_ok = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "repl", str(doc), "--backend", "pytorch"],
 +        )
 +        assert repl_ok.exit_code == 5, repl_ok.output
++
 +    def test_repl_error_mappings(self, tmp_path: Path, monkeypatch: Any) -> None:
 +        doc = _init_doc(tmp_path)
 +        runner = CliRunner()
++
 +        from dlm.base_models.errors import GatedModelError
 +        from dlm.inference import AdapterNotFoundError
 +        from dlm.inference.backends.select import UnsupportedBackendError
++
 +        original = doc.read_text(encoding="utf-8")
 +        fm_end = original.find("\n---\n", original.find("---") + 3)
 +        multi = tmp_path / "multi.dlm"
 +        multi.write_text(
 +            original[:fm_end] + "\ntraining:\n  adapters:\n    knowledge: {}\n" + original[fm_end:],
 +            encoding="utf-8",
 +        )
 +        unknown = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "repl", str(multi), "--adapter", "ghost"],
 +        )
 +        assert unknown.exit_code == 2, unknown.output
 +        assert "not declared" in unknown.output
++
 +        monkeypatch.setattr(
 +            "dlm.base_models.resolve",
 +            lambda *args, **kwargs: (_ for _ in ()).throw(
 +                GatedModelError("hf/model", "https://license")
 +            ),
 +        )
 +        gated = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "repl", str(doc), "--backend", "pytorch"],
 +        )
 +        assert gated.exit_code == 1, gated.output
 +        assert "run `dlm train --i-accept-license` first" in gated.output
++
 +        monkeypatch.setattr("dlm.base_models.resolve", lambda *args, **kwargs: SimpleNamespace())
 +        monkeypatch.setattr("dlm.hardware.doctor", lambda: SimpleNamespace(capabilities=object()))
 +        monkeypatch.setattr(
 +            "dlm.inference.backends.select_backend",
 +            lambda *args, **kwargs: (_ for _ in ()).throw(
 +                UnsupportedBackendError("backend not available")
 +            ),
 +        )
 +        unsupported = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "repl", str(doc), "--backend", "pytorch"],
 +        )
 +        assert unsupported.exit_code == 2, unsupported.output
 +        assert "backend not available" in unsupported.output
++
 +        class _MissingAdapterBackend:
 +            def load(self, spec: object, store: object, adapter_name: str | None = None) -> None:
 +                raise AdapterNotFoundError("missing adapter")
++
 +        monkeypatch.setattr(
 +            "dlm.inference.backends.select_backend", lambda *args, **kwargs: "pytorch"
 +        )
 +        monkeypatch.setattr(
 +            "dlm.inference.backends.build_backend",
 +            lambda *args, **kwargs: _MissingAdapterBackend(),
 +        )
 +        missing = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "repl", str(doc), "--backend", "pytorch"],
 +        )
 +        assert missing.exit_code == 1, missing.output
 +        assert "missing adapter" in missing.output
++
 +    def test_prompt_empty_query_and_repl_invalid_backend(
 +        self,
 +        tmp_path: Path,
 +        monkeypatch: Any,
 +    ) -> None:
 +        doc = _init_doc(tmp_path)
 +        runner = CliRunner()
++
 +        class _FakeBackend:
 +            def load(self, spec: object, store: object, adapter_name: str | None = None) -> None:
 +                return None
++
 +        monkeypatch.setattr("dlm.hardware.doctor", lambda: SimpleNamespace(capabilities=object()))
 +        monkeypatch.setattr(
 +            "dlm.inference.backends.select_backend", lambda *args, **kwargs: "pytorch"
 +        )
 +        monkeypatch.setattr(
 +            "dlm.inference.backends.build_backend", lambda *args, **kwargs: _FakeBackend()
 +        )
++
 +        prompt_result = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "prompt", str(doc)],
 +            input="",
 +        )
 +        assert prompt_result.exit_code == 2, prompt_result.output
 +        assert "empty query" in prompt_result.output
++
 +        repl_result = runner.invoke(
 +            app,
 +            ["--home", str(tmp_path / "home"), "repl", str(doc), "--backend", "bogus"],
 +        )
 +        assert repl_result.exit_code == 2, repl_result.output
 +        assert "--backend must be" in repl_result.output