diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py
index 3455267..7675043 100644
--- a/src/openultrasast/cli.py
+++ b/src/openultrasast/cli.py
@@ -3,7 +3,7 @@ from __future__ import annotations
 import argparse
 import json
 import tempfile
-from collections.abc import Sequence
+from collections.abc import Callable, Sequence
 from dataclasses import dataclass, replace
 from pathlib import Path
 from time import perf_counter
@@ -42,6 +42,7 @@ from .hunter import run_hunter_pool, write_hunter_trajectories
 from .hunter_harness import HxScanOrchestrator
 from .improve import RoundOutcome, run_improvement
 from .index import build_code_chunks
+from .learning.families import FamilyTaxonomy
 from .mapping import analyze_entry_points, attach_reachability_hints, ingest_sarif, write_entry_points, write_static_hints
 from .pair_gate import print_pair_metrics
 from .pairs import (
@@ -184,6 +185,7 @@ def main(argv: list[str] | None = None) -> int:
     )
     pairs.add_argument("--loo-out", type=Path, default=None, help="where to write loo.json (default: reports/loo.json)")
     pairs.add_argument("--json", action="store_true", help="print the pair scoreboard as JSON")
+    pairs.add_argument("--k-runs", type=int, default=1, help="runs per pair on the hunter path; the family scorer needs at least three")
 
     subparsers.add_parser("mcp", help="run the narrow MCP server over stdio for OpenCode integration")
     mechanisms = subparsers.add_parser("mechanisms", help="maintainer: mechanism memory seeded from trusted pairs")
@@ -199,6 +201,32 @@ def main(argv: list[str] | None = None) -> int:
     )
     export.add_argument("--json", action="store_true")
 
+    learning = subparsers.add_parser("learning", help="maintainer: the classified detector set and its rounds")
+    learning_sub = learning.add_subparsers(dest="learning_command", required=True)
+    for name, help_text in (
+        ("classify", "classify every labeled pair and report the classifier against itself"),
+        ("score", "score a catalog by family, with the denominator every number was computed over"),
+        ("baseline", "round zero: clone one detector into every family and measure each against itself"),
+        ("round", "one evolve round: propose one change for one family and let the evidence decide"),
+        ("publish", "regenerate every published number from the artifacts that produced it"),
+    ):
+        command = learning_sub.add_parser(name, help=help_text)
+        command.add_argument("--out", type=Path, default=Path(".openultrasast/learning"), help="where artifacts are read and written")
+        if name != "publish":
+            command.add_argument("--catalog", type=Path, default=DEFAULT_CATALOG)
+            command.add_argument("--slice", choices=SLICE_NAMES, default="all")
+            command.add_argument("--json", action="store_true")
+        if name in {"baseline", "round", "score"}:
+            command.add_argument("--k-runs", type=int, default=None, help="runs per pair; never fewer than three")
+            command.add_argument("--model", default=None, help="detector model; artifacts are keyed by it")
+            command.add_argument("--config", type=Path, default=Path("openultrasast.toml"))
+        if name == "round":
+            command.add_argument("--family", required=True)
+            command.add_argument("--cost-cap-usd", type=float, default=None)
+        if name == "publish":
+            command.add_argument("--measurements", type=Path, default=Path("benchmarks/measurements"))
+            command.add_argument("--roadmap", type=Path, default=Path(".kiro/steering/roadmap.md"))
+
     args = parser.parse_args(argv)
     if args.command == "scan":
         return _scan(args.path, args.config, args.mode, args.fail_on)
@@ -238,6 +266,8 @@ def main(argv: list[str] | None = None) -> int:
         )
     if args.command == "mechanisms":
         return _mechanisms_export(args.catalog, args.slice, args.store, json_out=args.json)
+    if args.command == "learning":
+        return _learning(args)
     if args.command == "mcp":
         from .mcp import serve  # lazy: keeps the import cycle (mcp -> cli) one-directional
 
@@ -1162,6 +1192,129 @@ def _run_family_detectors(
     return findings, payload, cited
 
 
+def _learning(args: argparse.Namespace) -> int:
+    """The maintainer's entry points into the classified detector set. Everything degrades with a reason."""
+    from .learning.classify import measure_classifier, write_review_queue
+    from .learning.families import load_families
+    from .learning.publish import publish
+
+    out: Path = args.out
+    taxonomy = load_families()
+    if args.learning_command == "publish":
+        report = publish(learning_dir=out, measurements_dir=args.measurements, roadmap=args.roadmap)
+        print(f"learning publish: {len(report.families)} families over {len(report.models)} model(s) -> {args.roadmap}")
+        for artifact in report.artifacts:
+            print(f"  artifact {artifact}")
+        return 0
+    cases = select_vendored(select_slice(load_pair_catalog(args.catalog), args.slice))
+    if args.learning_command == "classify":
+        measured = measure_classifier(cases, taxonomy, client=None)
+        out.mkdir(parents=True, exist_ok=True)
+        payload = measured.to_dict()
+        (out / "classifier.json").write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+        written = write_review_queue(out / "review-queue.jsonl", measured.queue)
+        if args.json:
+            print(json.dumps(payload, indent=2, sort_keys=True))
+        else:
+            print(f"learning classify: classified {measured.classified}, unknown {measured.unknown}, queued {written}")
+        return 0
+    return _learning_run(args, cases, taxonomy)
+
+
+def _learning_run(args: argparse.Namespace, cases: Sequence[PairCase], taxonomy: FamilyTaxonomy) -> int:
+    """score, baseline and round: the three that need a detector, and degrade with a reason without one."""
+    from .learning.detectors import load_family_configs, write_default_configs
+    from .learning.endpoint import resolve_chat_endpoint, resolve_models
+    from .learning.journal import Archive, LearningJournal
+    from .learning.proposer import HarnessXProposer
+    from .learning.rounds import MIN_K_RUNS, load_noise_floors, run_baseline, run_learning_round
+    from .learning.scoring import aggregate
+    from .tool_hunter import _SYSTEM_PROMPT
+
+    out: Path = args.out
+    config = load_config(args.config if args.config.exists() else None)
+    k_runs = args.k_runs or max(config.learning.k_runs, MIN_K_RUNS)
+    model = args.model or resolve_models(config)[0]
+    configs_dir = out / "configs"
+    if not configs_dir.exists():
+        write_default_configs(configs_dir, taxonomy, prompt=_SYSTEM_PROMPT, version="0")  # type: ignore[arg-type]
+    resolved = resolve_chat_endpoint(config)
+    if resolved is None:
+        print("learning: learning_endpoint_unavailable; scoring what can be scored without a detector")
+    client = resolved[0] if resolved else None
+    configs = load_family_configs(configs_dir, taxonomy)  # type: ignore[arg-type]
+
+    def factory(family_config: object) -> Callable[[Path], list[StaticFinding]]:
+        from .learning.detectors import Region, run_family_detector
+
+        def scan(root: Path) -> list[StaticFinding]:
+            if client is None:
+                return []
+            region = Region(path=_first_source(root), function=None)
+            return run_family_detector(root, region, family_config, client=client, model=model)  # type: ignore[arg-type]
+
+        return scan
+
+    if args.learning_command == "score":
+        from .learning.rounds import score_case
+
+        scores = [
+            score_case(case, factory(configs[_label(case)]), taxonomy=taxonomy, runs=k_runs) for case in cases if _label(case) in configs
+        ]
+        families = {name: block.to_dict() for name, block in aggregate(scores, taxonomy=taxonomy).items()}
+        if args.json:
+            print(json.dumps({"families": families}, indent=2, sort_keys=True))
+        else:
+            for name, block in sorted(families.items()):
+                print(f"learning score {name}: scorable={block['scorable']} recall={block['recall']:.3f} youden={block['youden']:.3f}")
+        return 0
+    if args.learning_command == "baseline":
+        report = run_baseline(
+            cases,
+            taxonomy=taxonomy,
+            configs_dir=configs_dir,
+            scan=factory(configs.get("unknown") or next(iter(configs.values()))),
+            model=model,
+            out_dir=out,
+            k_runs=k_runs,
+        )
+        print(f"learning baseline {model}: {len(report.floors)} families, k={report.k_runs} -> {out / 'baseline'}")
+        return 0
+    journal = LearningJournal(out / "journal.jsonl")
+    record = run_learning_round(
+        cases,
+        family=args.family,
+        taxonomy=taxonomy,
+        configs_dir=configs_dir,
+        proposer=HarnessXProposer(configs_dir=configs_dir, model=model),
+        scan_factory=factory,
+        model=model,
+        journal=journal,
+        archive=Archive(out / "archive.jsonl"),
+        floors=load_noise_floors(out, model),
+        out_dir=out,
+        cost_cap_usd=args.cost_cap_usd or config.learning.round_cost_cap_usd,
+        minibatch=config.learning.minibatch,
+        k_runs=k_runs,
+    )
+    print(f"learning round {record.round} {record.family}: {record.outcome} ({record.reason})")
+    return 0
+
+
+def _label(case: PairCase) -> str:
+    for row in case.expected:
+        if row.family:
+            return str(row.family)
+    return "unknown"
+
+
+def _first_source(root: Path) -> str:
+    from .preprocess import preprocess_repository
+
+    _, targets = preprocess_repository(root)
+    return targets[0].path if targets else "."
+
+
 def _hotspot_from_variant(finding: StaticFinding) -> Hotspot:
     return hotspot_from_finding(finding, rationale="variant of a known mechanism (suspicion; sandbox may raise)")
 
diff --git a/src/openultrasast/learning/rounds.py b/src/openultrasast/learning/rounds.py
index 563a549..a65cf9a 100644
--- a/src/openultrasast/learning/rounds.py
+++ b/src/openultrasast/learning/rounds.py
@@ -241,7 +241,7 @@ def run_learning_round(
     sweep_families: Sequence[str] = (),
     spent_usd: Callable[[], float] | None = None,
     trajectories: Callable[[], Sequence[Mapping[str, object]]] | None = None,
-) -> object:
+) -> RoundRecord:
     """One round: propose one change for one family, then let the evidence decide whether it stays.
 
     Staged so that the cheap answer comes first: a train minibatch, then the family's held-out pairs,
@@ -437,7 +437,7 @@ def _finish(
     scores: Sequence[PairFamilyScore] = (),
     version: str | None = None,
     trajectories: Callable[[], Sequence[Mapping[str, object]]] | None = None,
-) -> object:
+) -> RoundRecord:
     """Write the round directory and journal the outcome. Every outcome is journalled, not just the good ones."""
     from ..redaction import redact_secrets
 
diff --git a/tests/test_learning_cli.py b/tests/test_learning_cli.py
new file mode 100644
index 0000000..4fd9c78
--- /dev/null
+++ b/tests/test_learning_cli.py
@@ -0,0 +1,159 @@
+"""learning-harness task 3.9: the learning subcommands (offline)."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from openultrasast.cli import main
+
+VULN = "from flask import request\nimport os\n\n\ndef run():\n    os.system(request.args.get('cmd'))\n"
+FIXED = "from flask import request\nimport subprocess\n\n\ndef run():\n    subprocess.run(['echo', 'x'], check=False)\n"
+CATALOG = (
+    '[[pair]]\nname = "{name}"\nslice = "vibe-py"\nlanguage = "python"\nvuln = "{name}-v.py"\nfixed = "{name}-f.py"\n'
+    'relpath = "app.py"\nsplit = "{split}"\nreview_tier = "seeded"\n\n'
+    '[[pair.expected]]\ncwe = "CWE-78"\nclass = "command injection"\npath = "app.py"\nfunction = "run"\n'
+    'sink = "system"\nmechanism = "source_reaches_sink"\nfamily = "injection"\n'
+)
+
+
+@pytest.fixture
+def workspace(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    for name in ("train-a", "hold-a"):
+        (tmp_path / f"{name}-v.py").write_text(VULN)
+        (tmp_path / f"{name}-f.py").write_text(FIXED)
+    (tmp_path / "pairs.toml").write_text(
+        CATALOG.format(name="train-a", split="train") + "\n" + CATALOG.format(name="hold-a", split="holdout")
+    )
+    monkeypatch.setenv("OPENULTRASAST_HUNTER_CLIENT", "scripted")
+    return tmp_path  # no chdir: the mechanism vocabulary is found relative to the repository root
+
+
+def test_classify_reports_the_corpus_and_writes_the_queue(workspace: Path, capsys: pytest.CaptureFixture[str]) -> None:
+    assert main(["learning", "classify", "--catalog", str(workspace / "pairs.toml"), "--out", str(workspace / "learning")]) == 0
+    payload = json.loads((workspace / "learning" / "classifier.json").read_text())
+    assert payload["classified"] == 2 and payload["per_family"]["injection"] == 2
+    assert payload["hierarchical_credit"] is False
+    assert "classified 2" in capsys.readouterr().out
+
+
+def test_score_reports_family_numbers_for_a_catalog(workspace: Path, capsys: pytest.CaptureFixture[str]) -> None:
+    assert main(["learning", "score", "--catalog", str(workspace / "pairs.toml"), "--out", str(workspace / "learning"), "--json"]) == 0
+    payload = json.loads(capsys.readouterr().out)
+    assert "injection" in payload["families"]
+    assert payload["families"]["injection"]["taxonomy_version"]
+
+
+def test_baseline_writes_the_noise_floor_and_publish_reads_it(workspace: Path, capsys: pytest.CaptureFixture[str]) -> None:
+    assert (
+        main(
+            [
+                "learning",
+                "baseline",
+                "--catalog",
+                str(workspace / "pairs.toml"),
+                "--out",
+                str(workspace / "learning"),
+                "--k-runs",
+                "3",
+                "--model",
+                "scripted",
+            ]
+        )
+        == 0
+    )
+    floors = json.loads((workspace / "learning" / "baseline" / "scripted" / "noise-floors.json").read_text())
+    assert floors["k_runs"] == 3 and "injection" in floors["floors"]
+    capsys.readouterr()
+    roadmap = workspace / "roadmap.md"
+    roadmap.write_text("# Roadmap\n\n## Overview\n\ntext\n")
+    assert (
+        main(
+            ["learning", "publish", "--out", str(workspace / "learning"), "--measurements", str(workspace / "m"), "--roadmap", str(roadmap)]
+        )
+        == 0
+    )
+    assert "learning-harness:begin" in roadmap.read_text()
+    assert "| injection |" in roadmap.read_text()
+
+
+def test_a_round_runs_from_the_command_line_and_journals_its_outcome(workspace: Path) -> None:
+    assert (
+        main(
+            [
+                "learning",
+                "baseline",
+                "--catalog",
+                str(workspace / "pairs.toml"),
+                "--out",
+                str(workspace / "learning"),
+                "--k-runs",
+                "3",
+                "--model",
+                "scripted",
+            ]
+        )
+        == 0
+    )
+    assert (
+        main(
+            [
+                "learning",
+                "round",
+                "--catalog",
+                str(workspace / "pairs.toml"),
+                "--out",
+                str(workspace / "learning"),
+                "--family",
+                "injection",
+                "--k-runs",
+                "3",
+            ]
+        )
+        == 0
+    )
+    rounds = [json.loads(line) for line in (workspace / "learning" / "journal.jsonl").read_text().splitlines() if line.strip()]
+    assert len(rounds) == 1 and rounds[0]["family"] == "injection"
+    assert rounds[0]["outcome"] in {"accepted", "rejected", "reverted_cost"}
+    assert (workspace / "learning" / "rounds" / "1" / "attribution.json").is_file()
+
+
+def test_a_missing_endpoint_is_recorded_and_the_rest_still_runs(workspace: Path, monkeypatch: pytest.MonkeyPatch, capsys) -> None:  # type: ignore[no-untyped-def]
+    for name in ("OPENULTRASAST_HUNTER_CLIENT", "DEEPSEEK_API_KEY", "OPENROUTER_API_KEY"):
+        monkeypatch.delenv(name, raising=False)
+    assert (
+        main(
+            [
+                "learning",
+                "baseline",
+                "--catalog",
+                str(workspace / "pairs.toml"),
+                "--out",
+                str(workspace / "learning"),
+                "--k-runs",
+                "3",
+                "--model",
+                "none",
+            ]
+        )
+        == 0
+    )
+    out = capsys.readouterr().out
+    assert "learning_endpoint_unavailable" in out
+    assert (workspace / "learning" / "baseline" / "none" / "report.json").is_file()  # the run still recorded what it could
+
+
+def test_every_subcommand_is_listed_in_help(capsys: pytest.CaptureFixture[str]) -> None:
+    with pytest.raises(SystemExit):
+        main(["learning", "--help"])
+    out = capsys.readouterr().out
+    for name in ("classify", "score", "baseline", "round", "publish"):
+        assert name in out
+
+
+def test_the_pairs_command_takes_a_k_runs_flag(capsys: pytest.CaptureFixture[str]) -> None:
+    with pytest.raises(SystemExit):
+        main(["pairs", "--help"])
+    assert "--k-runs" in capsys.readouterr().out