SLM-418: write replay-preference corpus + run real preference train (DSH5-10, seventh slice)

scripts/build_replay_preference_corpus.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+"""SLM-418 (DSH5-10) seventh slice: write the bounded replay-preference
+corpus to real ``PreferencePair`` corpus files.
+
+    python -m scripts.build_replay_preference_corpus
+    python -m scripts.build_replay_preference_corpus --split train --out outputs/data/preference/replay_preference_train_pairs.jsonl
+
+Writes into this repo's existing preference-pairs corpus root
+(``outputs/data/preference/``) -- never a second corpus tree. This script
+only builds and writes the corpus; feed the train-split file into the
+existing ``slm preference train`` harness and measure the held-out split with
+``scripts.measure_replay_preference_held_out_benefit`` (see
+docs/design/dsh5-10-replay-preference-rows.md's reproducibility commands).
+This is fixture-scale wiring evidence, not a ship-readiness claim, and
+completes in well under a second -- far inside
+``slm_training.levers.MAX_RUN_MINUTES``.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from slm_training.harnesses.preference.replay_preference_corpus import (
+    DEFAULT_HELD_OUT_OUT_PATH,
+    DEFAULT_TRAIN_OUT_PATH,
+    write_replay_preference_corpus,
+)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--split",
+        choices=("train", "held_out", "both"),
+        default="both",
+        help="Which split to write (default: both, one file each).",
+    )
+    parser.add_argument(
+        "--train-out", type=Path, default=DEFAULT_TRAIN_OUT_PATH,
+        help="Output path for the train-split pairs file.",
+    )
+    parser.add_argument(
+        "--held-out-out", type=Path, default=DEFAULT_HELD_OUT_OUT_PATH,
+        help="Output path for the held-out-split pairs file.",
+    )
+    args = parser.parse_args(argv)
+
+    reports = []
+    if args.split in ("train", "both"):
+        reports.append(write_replay_preference_corpus(args.train_out, "train"))
+    if args.split in ("held_out", "both"):
+        reports.append(write_replay_preference_corpus(args.held_out_out, "held_out"))
+
+    print(json.dumps([report.to_dict() for report in reports], indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/measure_replay_preference_held_out_benefit.py

@@ -0,0 +1,62 @@
+#!/usr/bin/env python3
+"""SLM-418 (DSH5-10) seventh slice: honest held-out pairwise-preference
+benefit measurement for a real ``slm preference train`` run against the
+replay-preference corpus.
+
+    python -m scripts.measure_replay_preference_held_out_benefit \\
+        --baseline-checkpoint outputs/runs/<scratch-id>/last.pt \\
+        --held-out-pairs outputs/data/preference/replay_preference_held_out_pairs.jsonl \\
+        [--trained-checkpoint outputs/runs/<preference-id>/model.pt]
+
+Loads a baseline (pre-preference-training) checkpoint and, when given, a
+trained (post ``slm preference train``) checkpoint, measures each one's
+pairwise chosen>rejected preference accuracy on the held-out-split pairs, and
+reports an honest ``verdict`` -- ``benefit_observed_fixture_scale`` only when
+the trained checkpoint's accuracy strictly exceeds the baseline's, otherwise
+``no_benefit_fixture_scale``. This is fixture-scale wiring evidence over a
+handful of real pairs, never a certified or powered ship-readiness claim --
+see docs/design/dsh5-10-replay-preference-rows.md's "Seventh slice".
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from slm_training.harnesses.preference import load_pairs
+from slm_training.harnesses.preference.train import (
+    evaluate_replay_preference_held_out_benefit,
+)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--baseline-checkpoint", type=Path, required=True)
+    parser.add_argument("--trained-checkpoint", type=Path, default=None)
+    parser.add_argument("--held-out-pairs", type=Path, required=True)
+    parser.add_argument("--device", default="cpu")
+    parser.add_argument(
+        "--seed", type=int, default=0,
+        help="torch.manual_seed applied before each measurement, for reproducibility.",
+    )
+    parser.add_argument("--out", type=Path, default=None)
+    args = parser.parse_args(argv)
+
+    held_out_pairs = load_pairs(args.held_out_pairs)
+    report = evaluate_replay_preference_held_out_benefit(
+        baseline_checkpoint=args.baseline_checkpoint,
+        trained_checkpoint=args.trained_checkpoint,
+        held_out_pairs=held_out_pairs,
+        device=args.device,
+        seed=args.seed,
+    )
+    payload = json.dumps(report, indent=2, sort_keys=True)
+    print(payload)
+    if args.out:
+        args.out.write_text(payload + "\n", encoding="utf-8")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/slm_training/harnesses/preference/replay_preference_context_view_variants.py

@@ -466,13 +466,22 @@ class ReplayPreferenceSessionV1:
     ``group_id`` is the unit ``split_for_group`` partitions on: every row in
     one session shares one split (the issue's adversarial control,
     "conversation variants stay in one split").
+
+    ``trace`` (seventh slice) is the real ``ConversationTraceV1`` the session
+    was built from, when one exists -- ``None`` for the ``merge_success``
+    session, since no trace object exists for a merge decision (see
+    ``replay_preference.py``'s own module docstring). This lets a caller feed
+    a session's rows into ``preference_pairs_from_trace`` directly, matching
+    the sixth slice's own trace/direct-path conversion convention, without
+    re-deriving a trace from ``state_lookup`` alone.
     """
 
     group_id: str
     split: Split
     report: OperatorEventMemoryReportV1
     state_lookup: dict
     receipts: tuple[OperatorTurnReceiptV1, ...]
+    trace: ConversationTraceV1 | None = None
 
 
 #: Chain lengths exercised by the rollback-chain sessions -- directly the
@@ -506,6 +515,7 @@ def synthesize_bounded_session_corpus() -> tuple[ReplayPreferenceSessionV1, ...]
                 report=report,
                 state_lookup=state_lookup_from_trace(trace),
                 receipts=receipts_from_trace(trace),
+                trace=trace,
             )
         )
 
@@ -520,6 +530,7 @@ def synthesize_bounded_session_corpus() -> tuple[ReplayPreferenceSessionV1, ...]
             report=checkout_fork_report,
             state_lookup=state_lookup_from_trace(checkout_fork_trace),
             receipts=receipts_from_trace(checkout_fork_trace),
+            trace=checkout_fork_trace,
         )
     )
 
@@ -534,6 +545,7 @@ def synthesize_bounded_session_corpus() -> tuple[ReplayPreferenceSessionV1, ...]
             report=pronoun_report,
             state_lookup=state_lookup_from_trace(pronoun_trace),
             receipts=receipts_from_trace(pronoun_trace),
+            trace=pronoun_trace,
         )
     )
 

src/slm_training/harnesses/preference/replay_preference_corpus.py

@@ -0,0 +1,175 @@
+"""SLM-418 (DSH5-10) seventh slice: write the bounded replay-preference corpus
+to real ``PreferencePair`` corpus files, using this repo's existing
+conversion functions and corpus root -- no second corpus shape or shadow
+path.
+
+The sixth slice (v7) added :func:`~slm_training.dsl.operators.
+replay_preference.preference_pair_from_replay_row` and
+:func:`~slm_training.dsl.operators.replay_preference.
+preference_pairs_from_trace`, which convert an already-extracted
+``OperatorReplayPreferenceRowV1`` into this repo's existing ``PreferencePair``
+shape, but never wrote a pair to a corpus file or fed one into the real
+``slm preference train``/``train-local`` harness (see that slice's own
+"Explicitly still not attempted" note). This module closes exactly that gap
+for the bounded synthetic corpus
+(:func:`~slm_training.harnesses.preference.
+replay_preference_context_view_variants.synthesize_bounded_session_corpus`):
+
+* :func:`replay_preference_pairs_for_split` converts every row in every
+  session of one split (``"train"`` or ``"held_out"``) into a
+  ``PreferencePair``, reusing ``preference_pairs_from_trace`` for every
+  session that carries a real ``ConversationTraceV1`` and
+  ``preference_pair_from_replay_row`` directly for ``merge_success`` (whose
+  row is grounded on a ``BranchEditV1`` tip, not any one trace) -- the exact
+  same trace/direct-path convention the sixth slice's own tests exercise, now
+  applied uniformly across a whole corpus instead of one hand-built trace.
+* :func:`write_replay_preference_corpus` writes those pairs to this repo's
+  existing preference-pairs corpus root (``outputs/data/preference/`` -- the
+  same root ``slm preference build-pairs --out`` already writes into) via the
+  existing ``write_pairs`` writer. No second corpus tree is invented.
+
+Held-out pairs are written to their own file and are never mixed into the
+train file -- this repo's isolation/provenance law ("never fit training data
+to holdouts"). See docs/design/dsh5-10-replay-preference-rows.md's "Seventh
+slice" for what real training run and held-out measurement this corpus feeds,
+and for what is still explicitly out of scope (the DSH3-selected
+``TypedOperatorPolicyScorer``, the four-baseline comparison, CAP0/CAP1/CAP2
+retention).
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from pathlib import Path
+
+from slm_training.dsl.operators.replay_preference import (
+    preference_pair_from_replay_row,
+    preference_pairs_from_trace,
+)
+from slm_training.harness_core.versioning import build_version_stamp
+from slm_training.harnesses.preference import PreferencePair, write_pairs
+from slm_training.harnesses.preference.local_decisions import Split
+from slm_training.harnesses.preference.replay_preference_context_view_variants import (
+    ReplayPreferenceSessionV1,
+    synthesize_bounded_session_corpus,
+)
+
+#: Canonical preference-pairs corpus root this repo already uses (the same
+#: root ``slm preference build-pairs --out`` writes into) -- never a second
+#: corpus tree.
+DEFAULT_TRAIN_OUT_PATH = Path("outputs/data/preference/replay_preference_train_pairs.jsonl")
+DEFAULT_HELD_OUT_OUT_PATH = Path(
+    "outputs/data/preference/replay_preference_held_out_pairs.jsonl"
+)
+
+
+def replay_preference_pairs_for_split(
+    sessions: Sequence[ReplayPreferenceSessionV1], split: Split
+) -> tuple[tuple[PreferencePair, ...], tuple[dict, ...]]:
+    """Convert every row of every ``split`` session into a ``PreferencePair``.
+
+    Returns ``(pairs, skipped)``. Sessions carrying a real
+    ``ConversationTraceV1`` (every relation but ``merge_success``) go through
+    ``preference_pairs_from_trace`` unchanged; a session with no trace (only
+    ``merge_success`` today) converts each row directly via
+    ``preference_pair_from_replay_row``, honestly skipping -- never
+    fabricating a prompt for -- a row whose ``input_state_id`` is missing
+    from that session's own ``state_lookup``. This mirrors, at corpus scale,
+    the exact skip/direct-path convention the sixth slice's tests already
+    prove at single-trace scale.
+    """
+    pairs: list[PreferencePair] = []
+    skipped: list[dict] = []
+    for session in sessions:
+        if session.split != split:
+            continue
+        if session.trace is not None:
+            session_pairs, session_skipped = preference_pairs_from_trace(
+                session.trace, session.report.rows
+            )
+            pairs.extend(session_pairs)
+            skipped.extend(dict(entry) for entry in session_skipped)
+            continue
+        for row in session.report.rows:
+            node = session.state_lookup.get(row.input_state_id)
+            if node is None:
+                skipped.append(
+                    {
+                        "row": row.to_dict(),
+                        "reason": (
+                            f"input_state_id {row.input_state_id!r} not found in "
+                            f"session {session.group_id!r}'s state_lookup"
+                        ),
+                    }
+                )
+                continue
+            pairs.append(
+                preference_pair_from_replay_row(row, input_state_source=node.state.source)
+            )
+    return tuple(pairs), tuple(skipped)
+
+
+@dataclass(frozen=True)
+class ReplayPreferenceCorpusBuildReportV1:
+    """Honest counts for one written replay-preference corpus split file."""
+
+    split: Split
+    out_path: str
+    pair_count: int
+    skipped_count: int
+    session_count: int
+    row_count: int
+    skipped: tuple[dict, ...]
+    version_stamp: dict
+    schema: str = "replay_preference_corpus_build_report/v1"
+
+    def to_dict(self) -> dict:
+        return {
+            "schema": self.schema,
+            "split": self.split,
+            "out_path": self.out_path,
+            "pair_count": self.pair_count,
+            "skipped_count": self.skipped_count,
+            "session_count": self.session_count,
+            "row_count": self.row_count,
+            "skipped": list(self.skipped),
+            "version_stamp": self.version_stamp,
+        }
+
+
+def write_replay_preference_corpus(
+    out_path: Path | str,
+    split: Split,
+    *,
+    sessions: Sequence[ReplayPreferenceSessionV1] | None = None,
+) -> ReplayPreferenceCorpusBuildReportV1:
+    """Build the bounded synthetic corpus and write one ``split`` to ``out_path``.
+
+    Writes into this repo's existing preference-pairs corpus root via the
+    existing ``write_pairs`` writer -- never a second corpus tree. Raises
+    ``ValueError`` (fails closed) rather than writing an empty corpus file
+    when a split produces zero pairs.
+    """
+    resolved_sessions = (
+        tuple(sessions) if sessions is not None else synthesize_bounded_session_corpus()
+    )
+    pairs, skipped = replay_preference_pairs_for_split(resolved_sessions, split)
+    if not pairs:
+        raise ValueError(
+            f"replay-preference corpus build produced zero {split!r} pairs; "
+            "refusing to write an empty corpus file"
+        )
+    out = Path(out_path)
+    write_pairs(out, list(pairs))
+    split_sessions = [session for session in resolved_sessions if session.split == split]
+    return ReplayPreferenceCorpusBuildReportV1(
+        split=split,
+        out_path=str(out),
+        pair_count=len(pairs),
+        skipped_count=len(skipped),
+        session_count=len(split_sessions),
+        row_count=sum(len(session.report.rows) for session in split_sessions),
+        skipped=skipped,
+        version_stamp=build_version_stamp("harness.preference.replay_preference_corpus"),
+    )