diff --git a/etc/requirements-ml.txt b/etc/requirements-ml.txt
new file mode 100644
index 0000000000..dc284f7558
--- /dev/null
+++ b/etc/requirements-ml.txt
@@ -0,0 +1,18 @@
+# Dependencies for training and exporting the required phrase model.
+# Install with: pip install -r etc/requirements-ml.txt
+
+torch>=2.0
+transformers>=4.44
+sentencepiece>=0.2
+protobuf>=3.20
+accelerate>=0.33
+pytorch-crf>=0.7.2
+safetensors>=0.4
+
+# Used explicitly for DeBERTa-large training on a 16 GB GPU.
+bitsandbytes>=0.43
+
+# ONNX export and CPU verification.
+numpy>=1.24
+onnx>=1.16
+onnxruntime>=1.18
diff --git a/etc/scripts/dataset_pipeline/export_onnx.py b/etc/scripts/dataset_pipeline/export_onnx.py
new file mode 100644
index 0000000000..b06aaa1f64
--- /dev/null
+++ b/etc/scripts/dataset_pipeline/export_onnx.py
@@ -0,0 +1,233 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Export a trained required phrase tagger for CPU inference."""
+
+import hashlib
+import json
+import os
+from pathlib import Path
+from types import SimpleNamespace
+
+import click
+
+os.environ.setdefault("USE_TF", "0")
+
+
+def viterbi_decode(emissions, start_transitions, transitions, end_transitions):
+    """Return the best tag path for one sequence."""
+    sequence_length = emissions.shape[0]
+    score = start_transitions + emissions[0]
+    backpointers = []
+
+    for step in range(1, sequence_length):
+        candidates = score[:, None] + transitions
+        best_source = candidates.argmax(axis=0)
+        score = candidates.max(axis=0) + emissions[step]
+        backpointers.append(best_source)
+
+    score = score + end_transitions
+    best = int(score.argmax())
+    path = [best]
+    for sources in reversed(backpointers):
+        best = int(sources[best])
+        path.append(best)
+    path.reverse()
+    return path
+
+
+def sha256(path):
+    """Return the hexadecimal SHA256 digest of a file."""
+    digest = hashlib.sha256()
+    with open(path, "rb") as stream:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def build_emissions_module(tagger):
+    """Wrap the trained backbone and classifier for ONNX export."""
+    import torch.nn as nn
+
+    class EmissionsModule(nn.Module):
+        def __init__(self):
+            super().__init__()
+            self.backbone = tagger.backbone
+            self.classifier = tagger.classifier
+
+        def forward(self, input_ids, attention_mask):
+            hidden = self.backbone(
+                input_ids=input_ids,
+                attention_mask=attention_mask,
+            ).last_hidden_state
+            return self.classifier(hidden)
+
+    return EmissionsModule().eval()
+
+
+def load_tagger(model_dir, train_config):
+    """Rebuild a tagger and strictly load its saved weights."""
+    import torch
+    from safetensors.torch import load_file
+
+    from phrase_model import PhraseTagger
+
+    config = SimpleNamespace(
+        model_name=train_config["model_name"],
+        model_revision=train_config.get("model_revision"),
+        use_crf=train_config["use_crf"],
+        aux_ce_weight=0.0,
+        label_weights=[1.0] * len(train_config["labels"]),
+    )
+    tagger = PhraseTagger(config)
+
+    model_dir = Path(model_dir)
+    safetensors_file = model_dir / "model.safetensors"
+    pytorch_file = model_dir / "pytorch_model.bin"
+    if safetensors_file.exists():
+        state = load_file(str(safetensors_file))
+    elif pytorch_file.exists():
+        state = torch.load(pytorch_file, map_location="cpu", weights_only=True)
+    else:
+        raise FileNotFoundError(f"No model weights found in {model_dir}")
+
+    # Checkpoints created before class weights became non-persistent contain
+    # this training-only tensor.
+    state.pop("class_weights", None)
+    for name, tensor in state.items():
+        if not torch.isfinite(tensor).all():
+            raise ValueError(f"Checkpoint tensor {name!r} contains non-finite values")
+
+    tagger.load_state_dict(state, strict=True)
+    return tagger.eval()
+
+
+def check_viterbi_matches_crf(tagger, num_tags):
+    """Verify NumPy and pytorch-crf return identical paths."""
+    import torch
+
+    start = tagger.crf.start_transitions.detach().cpu().numpy()
+    transitions = tagger.crf.transitions.detach().cpu().numpy()
+    end = tagger.crf.end_transitions.detach().cpu().numpy()
+
+    emissions = torch.randn(3, 14, num_tags)
+    mask = torch.ones(3, 14, dtype=torch.bool)
+    crf_paths = tagger.crf.decode(emissions, mask=mask)
+    for row in range(emissions.size(0)):
+        numpy_path = viterbi_decode(emissions[row].numpy(), start, transitions, end)
+        if numpy_path != crf_paths[row]:
+            raise AssertionError("NumPy Viterbi disagrees with pytorch-crf decoding")
+
+    return start, transitions, end
+
+
+def export(model_dir, output_dir, opset):
+    """Export ONNX emissions, CRF transitions, and a checksum manifest."""
+    import numpy as np
+    import torch
+    from transformers import AutoTokenizer
+
+    model_dir = Path(model_dir)
+    output_dir = Path(output_dir)
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+    config_path = model_dir / "train_config.json"
+    if not config_path.exists():
+        raise FileNotFoundError(f"No train_config.json found in {model_dir}")
+    train_config = json.loads(config_path.read_text(encoding="utf-8"))
+    labels = train_config["labels"]
+    use_crf = train_config["use_crf"]
+
+    tagger = load_tagger(model_dir, train_config)
+    emissions_module = build_emissions_module(tagger)
+    tokenizer = AutoTokenizer.from_pretrained(str(model_dir), use_fast=True)
+
+    sample = tokenizer(
+        "Licensed under the Apache License Version 2.0",
+        return_tensors="pt",
+    )
+    inputs = sample["input_ids"], sample["attention_mask"]
+
+    onnx_path = output_dir / "model.onnx"
+    torch.onnx.export(
+        emissions_module,
+        inputs,
+        str(onnx_path),
+        input_names=["input_ids", "attention_mask"],
+        output_names=["emissions"],
+        dynamic_axes={
+            "input_ids": {0: "batch", 1: "sequence"},
+            "attention_mask": {0: "batch", 1: "sequence"},
+            "emissions": {0: "batch", 1: "sequence"},
+        },
+        opset_version=opset,
+        do_constant_folding=True,
+    )
+
+    manifest = {
+        "labels": labels,
+        "onnx_model": sha256(onnx_path),
+    }
+
+    if use_crf:
+        start, transitions, end = check_viterbi_matches_crf(tagger, len(labels))
+        transitions_path = output_dir / "crf_transitions.npz"
+        np.savez(
+            transitions_path,
+            start=start,
+            transitions=transitions,
+            end=end,
+        )
+        manifest["crf_transitions"] = sha256(transitions_path)
+
+    import onnxruntime
+
+    session = onnxruntime.InferenceSession(
+        str(onnx_path),
+        providers=["CPUExecutionProvider"],
+    )
+    feeds = {
+        "input_ids": sample["input_ids"].numpy(),
+        "attention_mask": sample["attention_mask"].numpy(),
+    }
+    onnx_emissions = session.run(["emissions"], feeds)[0]
+    torch_emissions = emissions_module(*inputs).detach().numpy()
+    if not np.allclose(onnx_emissions, torch_emissions, atol=1e-3):
+        raise AssertionError("ONNX emissions differ from PyTorch emissions")
+
+    manifest_path = output_dir / "manifest.json"
+    manifest_path.write_text(json.dumps(manifest, indent=2), encoding="utf-8")
+    click.echo(f"wrote {onnx_path}")
+    if use_crf:
+        click.echo(f"wrote {output_dir / 'crf_transitions.npz'}")
+    click.echo(f"wrote {manifest_path}")
+
+
+@click.command()
+@click.option(
+    "--model-dir",
+    required=True,
+    type=click.Path(exists=True, file_okay=False, path_type=Path),
+    help="Directory containing the trained model and train_config.json.",
+)
+@click.option(
+    "--output-dir",
+    default=None,
+    type=click.Path(file_okay=False, path_type=Path),
+    help="Output directory; defaults to the model directory.",
+)
+@click.option("--opset", default=14, type=int, show_default=True)
+def main(model_dir, output_dir, opset):
+    """Export a trained required phrase tagger to ONNX."""
+    try:
+        export(model_dir, output_dir or model_dir, opset)
+    except ImportError as error:
+        raise click.ClickException(
+            f"{error}; install scancode-required-phrases[training,onnx]"
+        ) from error
+
+
+if __name__ == "__main__":
+    main()
diff --git a/etc/scripts/dataset_pipeline/phrase_model.py b/etc/scripts/dataset_pipeline/phrase_model.py
new file mode 100644
index 0000000000..0858d58695
--- /dev/null
+++ b/etc/scripts/dataset_pipeline/phrase_model.py
@@ -0,0 +1,228 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""DeBERTa model and Trainer support for required phrase tagging."""
+
+import os
+
+os.environ.setdefault("USE_TF", "0")
+
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+from torch.optim import AdamW
+from torchcrf import CRF
+from transformers import AutoModel
+from transformers import Trainer
+
+from train_model import first_subword_positions
+from train_model import IGNORE_INDEX
+from train_model import LABELS
+
+
+class PhraseTagger(nn.Module):
+    """DeBERTa backbone with a word-level token classifier and optional CRF."""
+
+    def __init__(self, config):
+        super().__init__()
+        self.use_crf = config.use_crf
+        self.aux_ce_weight = config.aux_ce_weight
+        self.num_labels = len(LABELS)
+
+        self.backbone = AutoModel.from_pretrained(
+            config.model_name,
+            revision=config.model_revision,
+        ).float()
+        self.backbone.gradient_checkpointing_enable()
+        self.backbone.enable_input_require_grads()
+
+        hidden_size = self.backbone.config.hidden_size
+        dropout = getattr(self.backbone.config, "hidden_dropout_prob", 0.1)
+        self.dropout = nn.Dropout(dropout)
+        self.classifier = nn.Linear(hidden_size, self.num_labels)
+
+        if self.use_crf:
+            self.crf = CRF(self.num_labels, batch_first=True)
+
+        if self.aux_ce_weight > 0:
+            self.register_buffer(
+                "class_weights",
+                torch.tensor(config.label_weights, dtype=torch.float),
+                persistent=False,
+            )
+        else:
+            self.class_weights = None
+
+    def emissions(self, input_ids, attention_mask):
+        """Return per-subword label scores."""
+        hidden = self.backbone(
+            input_ids=input_ids,
+            attention_mask=attention_mask,
+        ).last_hidden_state
+        return self.classifier(self.dropout(hidden))
+
+    def token_cross_entropy(self, emissions, labels):
+        """Return weighted cross entropy over labeled subwords."""
+        return F.cross_entropy(
+            emissions.reshape(-1, self.num_labels),
+            labels.reshape(-1),
+            weight=self.class_weights,
+            ignore_index=IGNORE_INDEX,
+        )
+
+    def gather_words(self, emissions, labels):
+        """Pack first-subword emissions and labels into word-level sequences."""
+        batch, _, num_labels = emissions.shape
+        is_word = labels.ne(IGNORE_INDEX)
+        lengths = is_word.sum(dim=1)
+        width = int(lengths.max().item())
+
+        word_emissions = emissions.new_zeros((batch, width, num_labels))
+        crf_tags = labels.new_zeros((batch, width))
+        eval_tags = labels.new_full((batch, width), IGNORE_INDEX)
+        mask = torch.zeros((batch, width), dtype=torch.bool, device=emissions.device)
+
+        for row in range(batch):
+            positions = is_word[row].nonzero(as_tuple=True)[0]
+            count = positions.numel()
+            word_emissions[row, :count] = emissions[row, positions]
+            tags = labels[row, positions]
+            crf_tags[row, :count] = tags
+            eval_tags[row, :count] = tags
+            mask[row, :count] = True
+
+        return word_emissions, crf_tags, eval_tags, mask
+
+    def forward(self, input_ids, attention_mask, labels=None):
+        emissions = self.emissions(input_ids, attention_mask)
+        result = {}
+
+        if not self.use_crf:
+            if labels is not None:
+                result["loss"] = self.token_cross_entropy(emissions, labels)
+                result["word_labels"] = labels
+            if not self.training:
+                result["predictions"] = emissions.argmax(dim=-1)
+            return result
+
+        if labels is None:
+            raise ValueError("CRF head needs labels to locate words")
+
+        word_emissions, crf_tags, eval_tags, mask = self.gather_words(emissions, labels)
+        word_emissions = word_emissions.float()
+
+        log_likelihood = self.crf(word_emissions, crf_tags, mask=mask, reduction="mean")
+        loss = -log_likelihood
+        if self.aux_ce_weight > 0:
+            loss = loss + self.aux_ce_weight * self.token_cross_entropy(emissions, labels)
+
+        result["loss"] = loss
+        result["word_labels"] = eval_tags
+
+        if not self.training:
+            decoded = self.crf.decode(word_emissions, mask=mask)
+            result["predictions"] = self.pad_decoded(decoded, mask.size(1), emissions.device)
+
+        return result
+
+    def predict_words(self, input_ids, attention_mask, word_ids):
+        """Return one label ID per word for a single rule."""
+        positions = first_subword_positions(word_ids)
+        if not positions:
+            return []
+
+        emissions = self.emissions(input_ids, attention_mask)
+        word_emissions = emissions[0, positions].unsqueeze(0).float()
+        if not self.use_crf:
+            return word_emissions.argmax(dim=-1)[0].tolist()
+
+        mask = torch.ones(word_emissions.shape[:2], dtype=torch.bool, device=emissions.device)
+        return self.crf.decode(word_emissions, mask=mask)[0]
+
+    @staticmethod
+    def pad_decoded(decoded, width, device):
+        """Return variable-length decoded paths as a padded tensor."""
+        predictions = torch.full(
+            (len(decoded), width),
+            IGNORE_INDEX,
+            dtype=torch.long,
+            device=device,
+        )
+        for row, path in enumerate(decoded):
+            if path:
+                predictions[row, : len(path)] = torch.tensor(
+                    path,
+                    dtype=torch.long,
+                    device=device,
+                )
+        return predictions
+
+
+def build_optimizer(config, model):
+    """Return the configured AdamW optimizer with layer-wise learning rates."""
+    num_layers = model.backbone.config.num_hidden_layers
+    no_decay = ("bias", "LayerNorm.weight", "layer_norm.weight")
+
+    def rate_for(name):
+        if name.startswith("classifier") or name.startswith("crf"):
+            return config.head_lr
+        if ".encoder.layer." in name:
+            layer = int(name.split(".encoder.layer.")[1].split(".")[0])
+            return config.base_lr * (config.layer_decay ** (num_layers - layer))
+        return config.base_lr * (config.layer_decay ** (num_layers + 1))
+
+    groups = []
+    for name, parameter in model.named_parameters():
+        if not parameter.requires_grad:
+            continue
+        decay = 0.0 if any(part in name for part in no_decay) else config.weight_decay
+        groups.append(
+            {
+                "params": [parameter],
+                "lr": rate_for(name),
+                "weight_decay": decay,
+            }
+        )
+
+    optimizer_args = {
+        "lr": config.base_lr,
+        "eps": config.adam_epsilon,
+        "betas": (0.9, 0.999),
+    }
+    if config.optimizer == "adamw":
+        return AdamW(groups, **optimizer_args)
+
+    if config.optimizer == "adamw-8bit":
+        try:
+            from bitsandbytes.optim import AdamW8bit
+        except ImportError as error:
+            raise RuntimeError(
+                "adamw-8bit requires bitsandbytes; install the training dependencies"
+            ) from error
+        return AdamW8bit(groups, **optimizer_args)
+
+    raise ValueError(f"Unsupported optimizer: {config.optimizer}")
+
+
+class PhraseTrainer(Trainer):
+    """Trainer adapter for PhraseTagger output dictionaries."""
+
+    def compute_loss(self, model, inputs, return_outputs=False, **kwargs):
+        outputs = model(**inputs)
+        loss = outputs["loss"]
+        if not torch.isfinite(loss):
+            raise FloatingPointError("Training produced a non-finite loss")
+        return (loss, outputs) if return_outputs else loss
+
+    def prediction_step(self, model, inputs, prediction_loss_only, ignore_keys=None):
+        inputs = self._prepare_inputs(inputs)
+        with torch.no_grad():
+            outputs = model(**inputs)
+            loss = outputs.get("loss")
+            if loss is not None:
+                loss = loss.detach()
+        if prediction_loss_only:
+            return loss, None, None
+        return loss, outputs["predictions"], outputs["word_labels"]
diff --git a/etc/scripts/dataset_pipeline/test_export_onnx.py b/etc/scripts/dataset_pipeline/test_export_onnx.py
new file mode 100644
index 0000000000..523840b105
--- /dev/null
+++ b/etc/scripts/dataset_pipeline/test_export_onnx.py
@@ -0,0 +1,47 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from pathlib import Path
+import sys
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent))
+
+np = pytest.importorskip("numpy")
+
+from export_onnx import sha256
+from export_onnx import viterbi_decode
+
+
+def test_viterbi_with_zero_transitions_is_argmax():
+    emissions = np.array(
+        [
+            [0.1, 0.9, 0.0, 0.0, 0.0],
+            [0.7, 0.2, 0.0, 0.0, 0.0],
+            [0.0, 0.0, 0.0, 0.0, 0.9],
+        ]
+    )
+    transitions = np.zeros((5, 5))
+    edges = np.zeros(5)
+
+    assert viterbi_decode(emissions, edges, transitions, edges) == [1, 0, 4]
+
+
+def test_viterbi_obeys_transition_scores():
+    emissions = np.array([[0.0, 1.0], [1.0, 0.0]])
+    transitions = np.array([[0.0, 0.0], [-100.0, 0.0]])
+    edges = np.zeros(2)
+
+    path = viterbi_decode(emissions, edges, transitions, edges)
+
+    assert path[0] == path[1]
+
+
+def test_sha256_is_stable(tmp_path):
+    path = tmp_path / "model.onnx"
+    path.write_bytes(b"model")
+
+    assert sha256(path) == "9372c470eeadd5ecd9c3c74c2b3cb633f8e2f2fad799250a0f70d652b6b825e4"
diff --git a/etc/scripts/dataset_pipeline/test_phrase_model.py b/etc/scripts/dataset_pipeline/test_phrase_model.py
new file mode 100644
index 0000000000..fa85b16cd3
--- /dev/null
+++ b/etc/scripts/dataset_pipeline/test_phrase_model.py
@@ -0,0 +1,142 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import os
+from pathlib import Path
+import sys
+from types import SimpleNamespace
+
+import pytest
+
+os.environ.setdefault("USE_TF", "0")
+sys.path.insert(0, str(Path(__file__).parent))
+
+torch = pytest.importorskip("torch")
+pytest.importorskip("torchcrf")
+pytest.importorskip("transformers")
+
+import phrase_model as model_module
+from phrase_model import build_optimizer
+from phrase_model import PhraseTagger
+from train_model import LABEL2ID
+from train_model import LABELS
+
+
+class FakeBackbone(torch.nn.Module):
+    def __init__(self):
+        super().__init__()
+        self.config = SimpleNamespace(
+            hidden_size=4,
+            hidden_dropout_prob=0.1,
+            num_hidden_layers=1,
+        )
+        self.encoder = torch.nn.Module()
+        self.encoder.layer = torch.nn.ModuleList([torch.nn.Linear(4, 4)])
+        self.embeddings = torch.nn.Linear(4, 4)
+
+    def gradient_checkpointing_enable(self):
+        pass
+
+    def enable_input_require_grads(self):
+        pass
+
+
+@pytest.fixture
+def config():
+    return SimpleNamespace(
+        model_name="fake-model",
+        model_revision="revision",
+        use_crf=True,
+        aux_ce_weight=0.3,
+        label_weights=[1.0] * len(LABELS),
+        optimizer="adamw",
+        base_lr=2e-5,
+        head_lr=1e-4,
+        layer_decay=0.98,
+        weight_decay=0.01,
+        adam_epsilon=1e-6,
+    )
+
+
+def test_class_weights_are_not_saved(monkeypatch, config):
+    monkeypatch.setattr(
+        model_module.AutoModel,
+        "from_pretrained",
+        lambda *args, **kwargs: FakeBackbone(),
+    )
+
+    tagger = PhraseTagger(config)
+
+    assert tagger.class_weights is not None
+    assert "class_weights" not in tagger.state_dict()
+
+
+def test_model_revision_is_passed_to_the_backbone(monkeypatch, config):
+    calls = []
+
+    def from_pretrained(*args, **kwargs):
+        calls.append((args, kwargs))
+        return FakeBackbone()
+
+    monkeypatch.setattr(model_module.AutoModel, "from_pretrained", from_pretrained)
+    PhraseTagger(config)
+
+    assert calls == [(("fake-model",), {"revision": "revision"})]
+
+
+def test_build_optimizer_uses_explicit_adamw(monkeypatch, config):
+    monkeypatch.setattr(
+        model_module.AutoModel,
+        "from_pretrained",
+        lambda *args, **kwargs: FakeBackbone(),
+    )
+    tagger = PhraseTagger(config)
+
+    optimizer = build_optimizer(config, tagger)
+
+    assert isinstance(optimizer, torch.optim.AdamW)
+    learning_rates = {group["lr"] for group in optimizer.param_groups}
+    assert config.head_lr in learning_rates
+    assert any(rate < config.base_lr for rate in learning_rates)
+
+
+def make_crf_tagger():
+    from torchcrf import CRF
+
+    tagger = PhraseTagger.__new__(PhraseTagger)
+    torch.nn.Module.__init__(tagger)
+    tagger.use_crf = True
+    tagger.num_labels = len(LABELS)
+    tagger.crf = CRF(len(LABELS), batch_first=True)
+    with torch.no_grad():
+        for parameter in tagger.crf.parameters():
+            parameter.zero_()
+    return tagger
+
+
+def test_predict_words_uses_first_subwords():
+    tagger = make_crf_tagger()
+    emissions = torch.zeros((1, 5, len(LABELS)))
+    emissions[0, 1, LABEL2ID["B-REQ"]] = 9.0
+    emissions[0, 2, LABEL2ID["E-REQ"]] = 9.0
+    emissions[0, 3, LABEL2ID["S-REQ"]] = 9.0
+    tagger.emissions = lambda input_ids, attention_mask: emissions
+
+    input_ids = torch.zeros((1, 5), dtype=torch.long)
+    tags = tagger.predict_words(
+        input_ids,
+        input_ids,
+        [None, 0, 1, 1, None],
+    )
+
+    assert tags == [LABEL2ID["B-REQ"], LABEL2ID["E-REQ"]]
+
+
+def test_predict_words_does_not_run_for_an_empty_sequence():
+    tagger = make_crf_tagger()
+    tagger.emissions = lambda *args: pytest.fail("emissions should not be computed")
+    input_ids = torch.zeros((1, 2), dtype=torch.long)
+
+    assert tagger.predict_words(input_ids, input_ids, [None, None]) == []
diff --git a/etc/scripts/dataset_pipeline/test_train_model.py b/etc/scripts/dataset_pipeline/test_train_model.py
new file mode 100644
index 0000000000..888bff254a
--- /dev/null
+++ b/etc/scripts/dataset_pipeline/test_train_model.py
@@ -0,0 +1,337 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import json
+from pathlib import Path
+import sys
+
+from click.testing import CliRunner
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent))
+
+from train_model import align_labels
+from train_model import compute_metrics
+from train_model import Config
+from train_model import decode_row
+from train_model import extract_spans
+from train_model import first_subword_positions
+from train_model import IGNORE_INDEX
+from train_model import LABEL2ID
+from train_model import main
+from train_model import PhraseDataset
+from train_model import prepare_output_dir
+from train_model import serializable_config
+from train_model import sha256
+from train_model import validate_bioes
+from train_model import validate_config
+from train_model import validate_record
+from train_model import validate_saved_state
+from train_model import validate_splits
+
+
+class FakeEncoding(dict):
+    def __init__(self, word_ids):
+        super().__init__()
+        self._word_ids = word_ids
+        self["input_ids"] = [0] * len(word_ids)
+        self["attention_mask"] = [1] * len(word_ids)
+
+    def word_ids(self):
+        return self._word_ids
+
+
+class FakeTokenizer:
+    def __call__(self, tokens, max_length=512, **kwargs):
+        word_ids = [None, *range(len(tokens)), None]
+        return FakeEncoding(word_ids[:max_length])
+
+
+def make_record(identifier="mit_1.RULE", tokens=None, labels=None):
+    return {
+        "identifier": identifier,
+        "license_expression": "mit",
+        "rule_type": "is_license_notice",
+        "text": "MIT License terms apply",
+        "tokens": tokens or ["MIT", "License", "terms", "apply"],
+        "bioes_labels": labels or ["B-REQ", "E-REQ", "O", "O"],
+    }
+
+
+def write_jsonl(path, records):
+    path.write_text(
+        "".join(json.dumps(record) + "\n" for record in records),
+        encoding="utf-8",
+    )
+
+
+@pytest.mark.parametrize(
+    "labels",
+    [
+        ["O"],
+        ["S-REQ"],
+        ["B-REQ", "E-REQ"],
+        ["B-REQ", "I-REQ", "E-REQ", "O", "S-REQ"],
+    ],
+)
+def test_validate_bioes_accepts_valid_sequences(labels):
+    assert validate_bioes(labels) is None
+
+
+@pytest.mark.parametrize(
+    "labels",
+    [
+        [],
+        ["I-REQ"],
+        ["E-REQ"],
+        ["B-REQ", "O"],
+        ["O", "I-REQ"],
+        ["B-REQ", "I-REQ"],
+    ],
+)
+def test_validate_bioes_rejects_invalid_sequences(labels):
+    assert validate_bioes(labels)
+
+
+def test_validate_record_accepts_a_complete_record():
+    record = make_record()
+    assert validate_record(record, "train.jsonl", 1) is record
+
+
+@pytest.mark.parametrize(
+    "field_name",
+    ["identifier", "license_expression", "rule_type", "text", "tokens", "bioes_labels"],
+)
+def test_validate_record_rejects_a_missing_field(field_name):
+    record = make_record()
+    del record[field_name]
+    with pytest.raises(ValueError, match=field_name):
+        validate_record(record, "train.jsonl", 4)
+
+
+def test_validate_record_rejects_mismatched_tokens_and_labels():
+    record = make_record(labels=["S-REQ"])
+    with pytest.raises(ValueError, match="tokens and"):
+        validate_record(record, "train.jsonl", 2)
+
+
+def test_validate_record_rejects_unknown_labels():
+    record = make_record(labels=["B-REQ", "BAD", "O", "O"])
+    with pytest.raises(ValueError, match="unknown labels"):
+        validate_record(record, "train.jsonl", 2)
+
+
+def test_align_labels_keeps_only_first_subwords():
+    tokenizer = lambda tokens, **kwargs: FakeEncoding([None, 0, 1, 1, None])
+    encoding, truncated, cut_phrase = align_labels(
+        ["MIT", "License"],
+        ["B-REQ", "E-REQ"],
+        tokenizer,
+        512,
+    )
+    assert encoding["labels"] == [
+        IGNORE_INDEX,
+        LABEL2ID["B-REQ"],
+        LABEL2ID["E-REQ"],
+        IGNORE_INDEX,
+        IGNORE_INDEX,
+    ]
+    assert not truncated
+    assert not cut_phrase
+
+
+def test_align_labels_detects_a_phrase_cut_by_truncation():
+    encoding, truncated, cut_phrase = align_labels(
+        ["prefix", "GNU", "General", "Public", "License"],
+        ["O", "B-REQ", "I-REQ", "I-REQ", "E-REQ"],
+        FakeTokenizer(),
+        4,
+    )
+    assert encoding["labels"][-1] == LABEL2ID["I-REQ"]
+    assert truncated
+    assert cut_phrase
+
+
+def test_align_labels_allows_safe_truncation():
+    _, truncated, cut_phrase = align_labels(
+        ["MIT", "License", "terms", "apply"],
+        ["B-REQ", "E-REQ", "O", "O"],
+        FakeTokenizer(),
+        4,
+    )
+    assert truncated
+    assert not cut_phrase
+
+
+def test_phrase_dataset_skips_a_cut_phrase(tmp_path):
+    path = tmp_path / "train.jsonl"
+    write_jsonl(
+        path,
+        [
+            make_record(identifier="safe.RULE"),
+            make_record(
+                identifier="cut.RULE",
+                tokens=["prefix", "GNU", "General", "Public", "License"],
+                labels=["O", "B-REQ", "I-REQ", "I-REQ", "E-REQ"],
+            ),
+        ],
+    )
+
+    dataset = PhraseDataset(path, FakeTokenizer(), max_length=4)
+
+    assert dataset.identifiers == ["safe.RULE"]
+    assert dataset.truncated == 2
+    assert dataset.cut_phrases == 1
+
+
+def test_validate_splits_with_real_dataset_objects(tmp_path):
+    paths = {}
+    for name, identifier in (("train", "a.RULE"), ("validation", "b.RULE")):
+        path = tmp_path / f"{name}.jsonl"
+        write_jsonl(path, [make_record(identifier=identifier)])
+        paths[name] = PhraseDataset(path, FakeTokenizer(), 512)
+    validate_splits(paths)
+
+
+def test_validate_splits_rejects_duplicate_identifiers(tmp_path):
+    datasets = {}
+    for name in ("train", "validation"):
+        path = tmp_path / f"{name}.jsonl"
+        write_jsonl(path, [make_record(identifier="same.RULE")])
+        datasets[name] = PhraseDataset(path, FakeTokenizer(), 512)
+    with pytest.raises(ValueError, match="both train and validation"):
+        validate_splits(datasets)
+
+
+@pytest.mark.parametrize(
+    "tags, expected",
+    [
+        (["O", "B-REQ", "I-REQ", "E-REQ", "O"], {(1, 3)}),
+        (["O", "S-REQ", "O"], {(1, 1)}),
+        (["S-REQ", "O", "B-REQ", "E-REQ"], {(0, 0), (2, 3)}),
+        (["O", "O"], set()),
+    ],
+)
+def test_extract_spans(tags, expected):
+    assert extract_spans(tags) == expected
+
+
+def test_decode_row_drops_ignored_positions():
+    predicted, actual = decode_row(
+        [LABEL2ID["B-REQ"], 0, LABEL2ID["E-REQ"]],
+        [LABEL2ID["B-REQ"], IGNORE_INDEX, LABEL2ID["E-REQ"]],
+    )
+    assert predicted == ["B-REQ", "E-REQ"]
+    assert actual == ["B-REQ", "E-REQ"]
+
+
+def test_compute_metrics_scores_exact_spans():
+    predictions = [[LABEL2ID["B-REQ"], LABEL2ID["E-REQ"], LABEL2ID["O"]]]
+    labels = [[LABEL2ID["B-REQ"], LABEL2ID["E-REQ"], LABEL2ID["O"]]]
+    scores = compute_metrics((predictions, labels))
+    assert scores["f1"] == 1.0
+    assert scores["exact_match"] == 1.0
+    assert scores["predicted_spans"] == scores["gold_spans"] == 1
+
+
+def test_first_subword_positions_skips_specials_and_continuations():
+    assert first_subword_positions([None, 0, 1, 1, 2, None]) == [1, 2, 4]
+
+
+def test_sha256_is_stable(tmp_path):
+    path = tmp_path / "data.jsonl"
+    path.write_bytes(b"required phrase\n")
+    assert sha256(path) == "792616e2062f96efb6ae2f69e8637b834e2db74354eb4f51e78eda329038cc70"
+
+
+def test_serializable_config_converts_paths(tmp_path):
+    config = Config(data_dir=tmp_path / "data", output_dir=tmp_path / "model")
+    values = serializable_config(config)
+    assert values["data_dir"] == str(tmp_path / "data")
+    assert values["output_dir"] == str(tmp_path / "model")
+
+
+def test_validate_config_accepts_the_default_training_settings(tmp_path):
+    config = Config(data_dir=tmp_path / "data", output_dir=tmp_path / "model")
+    validate_config(config)
+
+
+def test_validate_config_rejects_invalid_settings(tmp_path):
+    config = Config(
+        data_dir=tmp_path / "data",
+        output_dir=tmp_path / "model",
+        aux_ce_weight=-1,
+    )
+    with pytest.raises(ValueError, match="cannot be negative"):
+        validate_config(config)
+
+
+def test_prepare_output_dir_refuses_to_overwrite_a_run(tmp_path):
+    output_dir = tmp_path / "model"
+    output_dir.mkdir()
+    (output_dir / "model.safetensors").write_bytes(b"model")
+
+    with pytest.raises(ValueError, match="not empty"):
+        prepare_output_dir(output_dir, resume=False)
+
+
+def test_prepare_output_dir_requires_a_checkpoint_to_resume(tmp_path):
+    with pytest.raises(ValueError, match="No checkpoint"):
+        prepare_output_dir(tmp_path / "model", resume=True)
+
+
+def test_validate_saved_state_accepts_matching_tensors(tmp_path):
+    torch = pytest.importorskip("torch")
+    safetensors = pytest.importorskip("safetensors.torch")
+
+    model = torch.nn.Linear(2, 1)
+    path = tmp_path / "model.safetensors"
+    safetensors.save_file(model.state_dict(), str(path))
+    validate_saved_state(model, path)
+
+
+def test_validate_saved_state_rejects_unexpected_keys(tmp_path):
+    torch = pytest.importorskip("torch")
+    safetensors = pytest.importorskip("safetensors.torch")
+
+    model = torch.nn.Linear(2, 1)
+    state = dict(model.state_dict())
+    state["unexpected"] = torch.ones(1)
+    path = tmp_path / "model.safetensors"
+    safetensors.save_file(state, str(path))
+    with pytest.raises(ValueError, match="key mismatch"):
+        validate_saved_state(model, path)
+
+
+def test_validate_saved_state_rejects_wrong_shapes(tmp_path):
+    torch = pytest.importorskip("torch")
+    safetensors = pytest.importorskip("safetensors.torch")
+
+    model = torch.nn.Linear(2, 1)
+    state = dict(model.state_dict())
+    state["weight"] = torch.ones((1, 3))
+    path = tmp_path / "model.safetensors"
+    safetensors.save_file(state, str(path))
+    with pytest.raises(ValueError, match="has shape"):
+        validate_saved_state(model, path)
+
+
+def test_validate_saved_state_rejects_non_finite_values(tmp_path):
+    torch = pytest.importorskip("torch")
+    safetensors = pytest.importorskip("safetensors.torch")
+
+    model = torch.nn.Linear(2, 1)
+    state = dict(model.state_dict())
+    state["weight"] = torch.full_like(state["weight"], float("nan"))
+    path = tmp_path / "model.safetensors"
+    safetensors.save_file(state, str(path))
+    with pytest.raises(ValueError, match="non-finite"):
+        validate_saved_state(model, path)
+
+
+def test_cli_requires_test_evaluation_for_isr(tmp_path):
+    result = CliRunner().invoke(main, ["--data-dir", str(tmp_path), "--with-isr"])
+    assert result.exit_code == 2
+    assert "--with-isr requires --evaluate-test" in result.output
diff --git a/etc/scripts/dataset_pipeline/train_model.py b/etc/scripts/dataset_pipeline/train_model.py
new file mode 100644
index 0000000000..0bb8c02a46
--- /dev/null
+++ b/etc/scripts/dataset_pipeline/train_model.py
@@ -0,0 +1,752 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Train a DeBERTa BIOES tagger for required phrase spans."""
+
+from dataclasses import asdict
+from dataclasses import dataclass
+from dataclasses import field
+import hashlib
+import importlib.metadata
+import inspect
+import json
+import os
+from pathlib import Path
+import platform
+import random
+
+import click
+
+os.environ.setdefault("USE_TF", "0")
+
+
+LABELS = ["O", "B-REQ", "I-REQ", "E-REQ", "S-REQ"]
+LABEL2ID = {label: index for index, label in enumerate(LABELS)}
+ID2LABEL = {index: label for index, label in enumerate(LABELS)}
+IGNORE_INDEX = -100
+
+MODEL_NAME = "microsoft/deberta-v3-large"
+MAX_LENGTH = 512
+RECORD_FIELDS = (
+    "identifier",
+    "license_expression",
+    "rule_type",
+    "text",
+    "tokens",
+    "bioes_labels",
+)
+START_LABELS = {"O", "B-REQ", "S-REQ"}
+END_LABELS = {"O", "E-REQ", "S-REQ"}
+VALID_TRANSITIONS = {
+    "O": {"O", "B-REQ", "S-REQ"},
+    "B-REQ": {"I-REQ", "E-REQ"},
+    "I-REQ": {"I-REQ", "E-REQ"},
+    "E-REQ": {"O", "B-REQ", "S-REQ"},
+    "S-REQ": {"O", "B-REQ", "S-REQ"},
+}
+
+
+@dataclass
+class Config:
+    """Settings for one training run."""
+
+    data_dir: Path
+    output_dir: Path
+    model_name: str = MODEL_NAME
+    model_revision: str | None = None
+    max_length: int = MAX_LENGTH
+
+    epochs: int = 8
+    batch_size: int = 1
+    grad_accum: int = 16
+    base_lr: float = 2e-5
+    head_lr: float = 1e-4
+    layer_decay: float = 0.98
+    weight_decay: float = 0.01
+    warmup_ratio: float = 0.1
+    max_grad_norm: float = 0.5
+    adam_epsilon: float = 1e-6
+    early_stopping_patience: int = 3
+    optimizer: str = "adamw"
+    precision: str = "fp32"
+
+    limit: int = 0
+    resume: bool = False
+    use_crf: bool = True
+    aux_ce_weight: float = 0.3
+    evaluate_test: bool = False
+    with_isr: bool = False
+    seed: int = 42
+
+    label_weights: list = field(default_factory=lambda: [1.0, 2.0, 1.5, 1.5, 2.0])
+
+
+def validate_config(config):
+    """Validate settings before loading the tokenizer or model."""
+    if config.optimizer not in {"adamw", "adamw-8bit"}:
+        raise ValueError(f"Unsupported optimizer: {config.optimizer}")
+    if config.precision not in {"fp32", "bf16"}:
+        raise ValueError(f"Unsupported precision: {config.precision}")
+    if config.epochs < 1 or config.batch_size < 1 or config.grad_accum < 1:
+        raise ValueError("Epochs, batch size, and gradient accumulation must be positive")
+    if config.base_lr <= 0 or config.head_lr <= 0:
+        raise ValueError("Learning rates must be positive")
+    if config.aux_ce_weight < 0:
+        raise ValueError("Auxiliary loss weight cannot be negative")
+    if len(config.label_weights) != len(LABELS):
+        raise ValueError(f"Expected {len(LABELS)} label weights")
+
+
+def prepare_output_dir(output_dir, resume):
+    """Create a new output directory or validate a resumable one."""
+    output_dir = Path(output_dir)
+    checkpoints = list(output_dir.glob("checkpoint-*")) if output_dir.exists() else []
+    if resume and not checkpoints:
+        raise ValueError(f"No checkpoint found in {output_dir}")
+    if not resume and output_dir.exists() and any(output_dir.iterdir()):
+        raise ValueError(f"Output directory is not empty: {output_dir}")
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+
+def set_seed(seed):
+    """Seed Python, NumPy, and PyTorch."""
+    import numpy as np
+    import torch
+
+    random.seed(seed)
+    np.random.seed(seed)
+    torch.manual_seed(seed)
+    torch.cuda.manual_seed_all(seed)
+
+
+def load_jsonl(path):
+    """Yield parsed records from a JSONL file."""
+    with open(path, encoding="utf-8") as lines:
+        for line_number, line in enumerate(lines, 1):
+            line = line.strip()
+            if line:
+                yield line_number, json.loads(line)
+
+
+def validate_bioes(labels):
+    """Return an error for an invalid BIOES sequence, or None."""
+    if not labels:
+        return "has no labels"
+    unknown = sorted(set(labels) - set(LABELS))
+    if unknown:
+        return f"contains unknown labels: {unknown}"
+    if labels[0] not in START_LABELS:
+        return f"starts with {labels[0]}"
+    for previous, current in zip(labels, labels[1:]):
+        if current not in VALID_TRANSITIONS[previous]:
+            return f"contains invalid transition {previous} -> {current}"
+    if labels[-1] not in END_LABELS:
+        return f"ends with {labels[-1]}"
+
+
+def validate_record(record, path, line_number):
+    """Validate one dataset record and return it."""
+    location = f"{path} line {line_number}"
+    for field_name in RECORD_FIELDS:
+        if field_name not in record:
+            raise ValueError(f"{location}: missing {field_name!r}")
+
+    identifier = record["identifier"]
+    if not identifier:
+        raise ValueError(f"{location}: empty identifier")
+    if not record["license_expression"]:
+        raise ValueError(f"{location} ({identifier}): empty license expression")
+    if not record["rule_type"]:
+        raise ValueError(f"{location} ({identifier}): empty rule type")
+
+    tokens = record["tokens"]
+    labels = record["bioes_labels"]
+    if not tokens:
+        raise ValueError(f"{location} ({identifier}): no tokens")
+    if len(tokens) != len(labels):
+        raise ValueError(
+            f"{location} ({identifier}): {len(tokens)} tokens and {len(labels)} labels"
+        )
+
+    unknown = sorted(set(labels) - set(LABELS))
+    if unknown:
+        raise ValueError(f"{location} ({identifier}): unknown labels: {unknown}")
+
+    error = validate_bioes(labels)
+    if error:
+        raise ValueError(f"{location} ({identifier}): {error}")
+    return record
+
+
+def align_labels(tokens, word_labels, tokenizer, max_length):
+    """Tokenize words and label only the first subword of each word."""
+    encoding = tokenizer(
+        tokens,
+        is_split_into_words=True,
+        truncation=True,
+        max_length=max_length,
+    )
+
+    word_ids = encoding.word_ids()
+    label_ids = []
+    previous_word = None
+    for word_id in word_ids:
+        if word_id is None:
+            label_ids.append(IGNORE_INDEX)
+        elif word_id != previous_word:
+            label_ids.append(LABEL2ID[word_labels[word_id]])
+        else:
+            label_ids.append(IGNORE_INDEX)
+        previous_word = word_id
+
+    encoding["labels"] = label_ids
+    kept_word_ids = [word_id for word_id in word_ids if word_id is not None]
+    kept_words = max(kept_word_ids) + 1 if kept_word_ids else 0
+    truncated = kept_words < len(tokens)
+    cut_phrase = (
+        truncated
+        and kept_words
+        and word_labels[kept_words - 1]
+        in {
+            "B-REQ",
+            "I-REQ",
+        }
+    )
+    return encoding, truncated, bool(cut_phrase)
+
+
+def first_subword_positions(word_ids):
+    """Return positions that start a tokenizer word."""
+    positions = []
+    previous = None
+    for index, word_id in enumerate(word_ids):
+        if word_id is None:
+            previous = None
+            continue
+        if word_id != previous:
+            positions.append(index)
+        previous = word_id
+    return positions
+
+
+class PhraseDataset:
+    """Read and encode one BIOES JSONL split."""
+
+    def __init__(self, path, tokenizer, max_length, limit=0):
+        self.examples = []
+        self.identifiers = []
+        self.truncated = 0
+        self.cut_phrases = 0
+
+        for line_number, unvalidated_record in load_jsonl(path):
+            if limit and len(self.examples) >= limit:
+                break
+            record = validate_record(unvalidated_record, path, line_number)
+            encoding, truncated, cut_phrase = align_labels(
+                record["tokens"],
+                record["bioes_labels"],
+                tokenizer,
+                max_length,
+            )
+            if truncated:
+                self.truncated += 1
+            if cut_phrase:
+                self.cut_phrases += 1
+                continue
+
+            self.identifiers.append(record["identifier"])
+            self.examples.append(
+                {
+                    "input_ids": encoding["input_ids"],
+                    "attention_mask": encoding["attention_mask"],
+                    "labels": encoding["labels"],
+                }
+            )
+
+    def __len__(self):
+        return len(self.examples)
+
+    def __getitem__(self, index):
+        return self.examples[index]
+
+
+def validate_splits(datasets):
+    """Require non-empty splits with disjoint rule identifiers."""
+    seen = {}
+    for split_name, dataset in datasets.items():
+        if not dataset:
+            raise ValueError(f"{split_name} split has no usable examples")
+        for identifier in dataset.identifiers:
+            previous_split = seen.get(identifier)
+            if previous_split:
+                raise ValueError(
+                    f"Rule {identifier!r} occurs in both {previous_split} and {split_name}"
+                )
+            seen[identifier] = split_name
+
+
+def extract_spans(tags):
+    """Return the set of inclusive word spans in a BIOES sequence."""
+    spans = []
+    start = None
+    for index, tag in enumerate(tags):
+        if tag == "S-REQ":
+            spans.append((index, index))
+            start = None
+        elif tag == "B-REQ":
+            if start is not None:
+                spans.append((start, index - 1))
+            start = index
+        elif tag == "I-REQ":
+            if start is None:
+                start = index
+        elif tag == "E-REQ":
+            if start is None:
+                start = index
+            spans.append((start, index))
+            start = None
+        elif start is not None:
+            spans.append((start, index - 1))
+            start = None
+    if start is not None:
+        spans.append((start, len(tags) - 1))
+    return set(spans)
+
+
+def decode_row(pred_row, label_row):
+    """Drop ignored positions and map label IDs to BIOES tags."""
+    predicted = []
+    actual = []
+    for prediction, label in zip(pred_row, label_row):
+        if int(label) == IGNORE_INDEX:
+            continue
+        actual.append(ID2LABEL[int(label)])
+        predicted.append(ID2LABEL.get(int(prediction), "O"))
+    return predicted, actual
+
+
+def compute_metrics(eval_pred):
+    """Return strict span-level micro metrics and rule-level exact match."""
+    predictions, labels = eval_pred
+    true_positive = false_positive = false_negative = 0
+    exact = 0
+
+    for pred_row, label_row in zip(predictions, labels):
+        predicted, actual = decode_row(pred_row, label_row)
+        predicted_spans = extract_spans(predicted)
+        actual_spans = extract_spans(actual)
+        true_positive += len(predicted_spans & actual_spans)
+        false_positive += len(predicted_spans - actual_spans)
+        false_negative += len(actual_spans - predicted_spans)
+        if predicted_spans == actual_spans:
+            exact += 1
+
+    precision_denominator = true_positive + false_positive
+    recall_denominator = true_positive + false_negative
+    precision = true_positive / precision_denominator if precision_denominator else 0.0
+    recall = true_positive / recall_denominator if recall_denominator else 0.0
+    f1 = 2 * precision * recall / (precision + recall) if precision + recall else 0.0
+    return {
+        "f1": f1,
+        "precision": precision,
+        "recall": recall,
+        "exact_match": exact / len(predictions) if len(predictions) else 0.0,
+        "predicted_spans": true_positive + false_positive,
+        "gold_spans": true_positive + false_negative,
+    }
+
+
+def evaluate_isr(records, model, tokenizer, max_length):
+    """Return the fraction of predicted phrases ScanCode can locate."""
+    import torch
+    from licensedcode.required_phrases import find_phrase_spans_in_text
+
+    device = next(model.parameters()).device
+    model.eval()
+    total = 0
+    injectable = 0
+    for record in records:
+        encoding, _, cut_phrase = align_labels(
+            record["tokens"],
+            record["bioes_labels"],
+            tokenizer,
+            max_length,
+        )
+        if cut_phrase:
+            continue
+        inputs = {
+            "input_ids": torch.tensor([encoding["input_ids"]], device=device),
+            "attention_mask": torch.tensor([encoding["attention_mask"]], device=device),
+            "labels": torch.tensor([encoding["labels"]], device=device),
+        }
+        with torch.no_grad():
+            output = model(**inputs)
+        tags, _ = decode_row(
+            output["predictions"][0].tolist(),
+            output["word_labels"][0].tolist(),
+        )
+        for start, end in extract_spans(tags):
+            if end >= len(record["tokens"]):
+                continue
+            phrase = " ".join(record["tokens"][start : end + 1])
+            total += 1
+            if find_phrase_spans_in_text(record["text"], phrase):
+                injectable += 1
+
+    return injectable / total if total else 0.0
+
+
+def sha256(path):
+    """Return the hexadecimal SHA256 digest of a file."""
+    digest = hashlib.sha256()
+    with open(path, "rb") as stream:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def installed_version(package_name):
+    """Return an installed package version, or None."""
+    try:
+        return importlib.metadata.version(package_name)
+    except importlib.metadata.PackageNotFoundError:
+        return None
+
+
+def serializable_config(config):
+    """Return the training configuration with paths converted to strings."""
+    values = asdict(config)
+    values["data_dir"] = str(values["data_dir"])
+    values["output_dir"] = str(values["output_dir"])
+    return values
+
+
+def validate_saved_state(model, model_path):
+    """Validate the keys, shapes, and values in a saved safetensors model."""
+    import torch
+    from safetensors.torch import load_file
+
+    saved = load_file(str(model_path))
+    expected = model.state_dict()
+    if set(saved) != set(expected):
+        missing = sorted(set(expected) - set(saved))
+        unexpected = sorted(set(saved) - set(expected))
+        raise ValueError(f"Saved model key mismatch; missing={missing}, unexpected={unexpected}")
+
+    for name, tensor in saved.items():
+        if tensor.shape != expected[name].shape:
+            raise ValueError(
+                f"Saved tensor {name!r} has shape {tuple(tensor.shape)}, "
+                f"expected {tuple(expected[name].shape)}"
+            )
+        if not torch.isfinite(tensor).all():
+            raise ValueError(f"Saved tensor {name!r} contains non-finite values")
+
+
+def validate_precision(precision):
+    """Validate the selected training precision."""
+    import torch
+
+    if precision == "bf16" and not (torch.cuda.is_available() and torch.cuda.is_bf16_supported()):
+        raise ValueError("bf16 requires a CUDA device with BF16 support")
+
+
+def run_training(config):
+    """Train, validate, and save a required phrase model."""
+    import torch
+    import transformers
+    from transformers import AutoTokenizer
+    from transformers import DataCollatorForTokenClassification
+    from transformers import EarlyStoppingCallback
+    from transformers import TrainingArguments
+
+    from phrase_model import PhraseTagger
+    from phrase_model import PhraseTrainer
+    from phrase_model import build_optimizer
+
+    if config.with_isr and not config.evaluate_test:
+        raise ValueError("ISR evaluation requires --evaluate-test")
+
+    validate_config(config)
+    validate_precision(config.precision)
+    prepare_output_dir(config.output_dir, config.resume)
+    set_seed(config.seed)
+
+    tokenizer = AutoTokenizer.from_pretrained(
+        config.model_name,
+        revision=config.model_revision,
+        use_fast=True,
+    )
+    if not tokenizer.is_fast:
+        raise RuntimeError("Training requires a fast tokenizer with word IDs")
+
+    paths = {
+        "train": config.data_dir / "train.jsonl",
+        "validation": config.data_dir / "val.jsonl",
+        "test": config.data_dir / "test.jsonl",
+    }
+    datasets = {
+        name: PhraseDataset(path, tokenizer, config.max_length, config.limit)
+        for name, path in paths.items()
+    }
+    validate_splits(datasets)
+
+    for name, dataset in datasets.items():
+        click.echo(
+            f"{name}: {len(dataset)} examples, {dataset.truncated} truncated, "
+            f"{dataset.cut_phrases} skipped with a cut phrase"
+        )
+
+    model = PhraseTagger(config)
+    collator = DataCollatorForTokenClassification(
+        tokenizer,
+        label_pad_token_id=IGNORE_INDEX,
+    )
+
+    arguments = TrainingArguments(
+        output_dir=str(config.output_dir),
+        num_train_epochs=config.epochs,
+        per_device_train_batch_size=config.batch_size,
+        per_device_eval_batch_size=config.batch_size,
+        gradient_accumulation_steps=config.grad_accum,
+        learning_rate=config.base_lr,
+        weight_decay=config.weight_decay,
+        warmup_ratio=config.warmup_ratio,
+        lr_scheduler_type="cosine",
+        max_grad_norm=config.max_grad_norm,
+        eval_strategy="epoch",
+        save_strategy="epoch",
+        save_total_limit=2,
+        load_best_model_at_end=True,
+        metric_for_best_model="f1",
+        greater_is_better=True,
+        bf16=config.precision == "bf16",
+        fp16=False,
+        logging_steps=50,
+        report_to="none",
+        seed=config.seed,
+        data_seed=config.seed,
+        dataloader_num_workers=2,
+        save_safetensors=True,
+    )
+
+    trainer_kwargs = {
+        "model": model,
+        "args": arguments,
+        "train_dataset": datasets["train"],
+        "eval_dataset": datasets["validation"],
+        "data_collator": collator,
+        "compute_metrics": compute_metrics,
+        "optimizers": (build_optimizer(config, model), None),
+        "callbacks": [
+            EarlyStoppingCallback(
+                early_stopping_patience=config.early_stopping_patience,
+            )
+        ],
+    }
+    if "processing_class" in inspect.signature(PhraseTrainer.__init__).parameters:
+        trainer_kwargs["processing_class"] = tokenizer
+    else:
+        trainer_kwargs["tokenizer"] = tokenizer
+
+    trainer = PhraseTrainer(**trainer_kwargs)
+    trainer.train(resume_from_checkpoint=config.resume or None)
+
+    trainer.save_model(str(config.output_dir))
+    tokenizer.save_pretrained(str(config.output_dir))
+
+    resolved_revision = getattr(model.backbone.config, "_commit_hash", None)
+    train_config = {
+        "model_name": config.model_name,
+        "model_revision": resolved_revision or config.model_revision,
+        "use_crf": config.use_crf,
+        "max_length": config.max_length,
+        "labels": LABELS,
+    }
+    (config.output_dir / "train_config.json").write_text(
+        json.dumps(train_config, indent=2),
+        encoding="utf-8",
+    )
+
+    model_path = config.output_dir / "model.safetensors"
+    validate_saved_state(model, model_path)
+
+    validation_metrics = trainer.evaluate(
+        datasets["validation"],
+        metric_key_prefix="validation",
+    )
+    test_metrics = None
+    if config.evaluate_test:
+        test_metrics = trainer.evaluate(
+            datasets["test"],
+            metric_key_prefix="test",
+        )
+        if config.with_isr:
+            test_records = [
+                validate_record(record, paths["test"], line_number)
+                for line_number, record in load_jsonl(paths["test"])
+            ]
+            test_metrics["test_isr"] = evaluate_isr(
+                test_records,
+                model,
+                tokenizer,
+                config.max_length,
+            )
+
+    manifest = {
+        "config": serializable_config(config),
+        "dataset": {
+            name: {
+                "path": str(path),
+                "sha256": sha256(path),
+                "examples": len(datasets[name]),
+                "truncated": datasets[name].truncated,
+                "cut_phrases": datasets[name].cut_phrases,
+            }
+            for name, path in paths.items()
+        },
+        "model_revision": resolved_revision or config.model_revision,
+        "best_checkpoint": trainer.state.best_model_checkpoint,
+        "best_validation_f1": trainer.state.best_metric,
+        "validation_metrics": validation_metrics,
+        "test_metrics": test_metrics,
+        "log_history": trainer.state.log_history,
+        "versions": {
+            "python": platform.python_version(),
+            "torch": torch.__version__,
+            "transformers": transformers.__version__,
+            "scancode_toolkit": installed_version("scancode-toolkit"),
+            "pytorch_crf": installed_version("pytorch-crf"),
+        },
+        "gpu": torch.cuda.get_device_name(0) if torch.cuda.is_available() else None,
+    }
+    (config.output_dir / "run_manifest.json").write_text(
+        json.dumps(manifest, indent=2),
+        encoding="utf-8",
+    )
+
+    click.echo(f"best checkpoint: {trainer.state.best_model_checkpoint}")
+    click.echo(f"best validation F1: {trainer.state.best_metric}")
+    click.echo(f"validation: {validation_metrics}")
+    if test_metrics is not None:
+        click.echo(f"test: {test_metrics}")
+
+    return {
+        "validation": validation_metrics,
+        "test": test_metrics,
+    }
+
+
+@click.command()
+@click.option(
+    "--data-dir",
+    required=True,
+    type=click.Path(exists=True, file_okay=False, path_type=Path),
+    help="Directory containing train.jsonl, val.jsonl, and test.jsonl.",
+)
+@click.option(
+    "--output-dir",
+    default="model-output",
+    type=click.Path(file_okay=False, path_type=Path),
+    help="Directory for checkpoints and the final model.",
+)
+@click.option("--model-name", default=MODEL_NAME, help="Base model to fine-tune.")
+@click.option("--model-revision", default=None, help="Optional model revision to pin.")
+@click.option(
+    "--max-length",
+    default=MAX_LENGTH,
+    type=click.IntRange(min=3, max=MAX_LENGTH),
+    show_default=True,
+)
+@click.option("--epochs", default=8, type=click.IntRange(min=1), show_default=True)
+@click.option("--batch-size", default=1, type=int, show_default=True)
+@click.option("--grad-accum", default=16, type=int, show_default=True)
+@click.option("--base-lr", default=2e-5, type=float, show_default=True)
+@click.option("--head-lr", default=1e-4, type=float, show_default=True)
+@click.option("--aux-ce-weight", default=0.3, type=float, show_default=True)
+@click.option(
+    "--optimizer",
+    type=click.Choice(["adamw", "adamw-8bit"]),
+    default="adamw",
+    show_default=True,
+)
+@click.option(
+    "--precision",
+    type=click.Choice(["fp32", "bf16"]),
+    default="fp32",
+    show_default=True,
+)
+@click.option("--no-crf", is_flag=True, default=False, help="Train without the CRF head.")
+@click.option(
+    "--evaluate-test",
+    is_flag=True,
+    default=False,
+    help="Evaluate the test split after training.",
+)
+@click.option(
+    "--with-isr",
+    is_flag=True,
+    default=False,
+    help="Report injection success rate with final test evaluation.",
+)
+@click.option("--limit", default=0, type=int, help="Limit examples per split; zero uses all.")
+@click.option("--resume", is_flag=True, default=False, help="Resume from the latest checkpoint.")
+@click.option("--seed", default=42, type=int, show_default=True)
+def main(
+    data_dir,
+    output_dir,
+    model_name,
+    model_revision,
+    max_length,
+    epochs,
+    batch_size,
+    grad_accum,
+    base_lr,
+    head_lr,
+    aux_ce_weight,
+    optimizer,
+    precision,
+    no_crf,
+    evaluate_test,
+    with_isr,
+    limit,
+    resume,
+    seed,
+):
+    """Train the required phrase tagger from a BIOES dataset."""
+    if with_isr and not evaluate_test:
+        raise click.UsageError("--with-isr requires --evaluate-test")
+
+    config = Config(
+        data_dir=data_dir,
+        output_dir=output_dir,
+        model_name=model_name,
+        model_revision=model_revision,
+        max_length=max_length,
+        epochs=epochs,
+        batch_size=batch_size,
+        grad_accum=grad_accum,
+        base_lr=base_lr,
+        head_lr=head_lr,
+        aux_ce_weight=aux_ce_weight,
+        optimizer=optimizer,
+        precision=precision,
+        use_crf=not no_crf,
+        evaluate_test=evaluate_test,
+        with_isr=with_isr,
+        limit=limit,
+        resume=resume,
+        seed=seed,
+    )
+    try:
+        run_training(config)
+    except ImportError as error:
+        raise click.ClickException(
+            f"{error}; install scancode-required-phrases[training]"
+        ) from error
+
+
+if __name__ == "__main__":
+    main()