diff --git a/scripts/migrate_decisions_split.py b/scripts/migrate_decisions_split.py
new file mode 100644
index 000000000..7699c82a1
--- /dev/null
+++ b/scripts/migrate_decisions_split.py
@@ -0,0 +1,163 @@
+#!/usr/bin/env python3
+"""One-shot migration (ersatztv#610): split the decision corpus into one file per record.
+
+ docs/decisions.md -> docs/decisions/records//.md (one per record)
+ docs/decisions/.md + the source file KEPT, stripped to its narrative
+ docs/decisions/archive/*.md -> docs/decisions/archive//.md
+
+Run from the repo root: PYTHONPATH=. python3 scripts/migrate_decisions_split.py [--dry-run]
+
+Two properties this script exists to guarantee, both checked by
+`scripts/tests/test_migration_equivalence.py`:
+
+1. **Nothing is rewritten.** Every record's rationale body is copied byte-for-byte, and every
+ metadata field round-trips unchanged. The migration is a MOVE, so a field-level before/after
+ comparison is a complete correctness proof.
+2. **No prose is lost.** 506 lines of the corpus are NOT inside any record — file preambles that
+ explain why a topic file exists, cross-reference notes, and decisions.md's lifecycle schema
+ header. Those files are kept and stripped to exactly that prose, never deleted. (They cannot be
+ filed per-area: topic files hold several areas, and 4 of 23 areas span several files.)
+"""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+import yaml
+
+import scripts.decisions_lib as dl
+
+# Emitted in this order so every record file reads the same way.
+_FIELD_ORDER = [
+ ("key", "key"),
+ ("title", None), # from Record.heading
+ ("status", "status"),
+ ("since", "since"),
+ ("supersedes", "supersedes"),
+ ("superseded-by", "superseded_by"),
+ ("stale-after", "stale_after"),
+ ("rule", "rule"),
+ ("signals", "signals"),
+ ("mechanics", "mechanics"),
+ ("sources", "sources"),
+]
+
+
+def record_path(rec: dl.Record) -> Path:
+ """`records//.md` for an active record, `archive//.md` otherwise.
+
+ Identity stays the KEY; the path is derived from it, and the validator enforces the two agree.
+ Every key in the corpus is exactly two dotted segments, so this is unambiguous — a key with
+ inner dots would need a rule and there are none.
+ """
+ area, _, topic = (rec.key or "").partition(".")
+ root = dl.ARCHIVE_DIR if rec.status in ("superseded", "retired") else dl.RECORDS_DIR
+ return root / area / f"{topic}.md"
+
+
+def render_record(rec: dl.Record) -> str:
+ """YAML frontmatter + the rationale body, copied verbatim."""
+ meta: dict[str, object] = {}
+ for fm_key, field in _FIELD_ORDER:
+ val = rec.heading if fm_key == "title" else getattr(rec, field)
+ if val is None:
+ continue # absent stays absent; "" is emitted so empty-vs-absent survives
+ meta[fm_key] = val
+ fm = yaml.safe_dump(
+ meta,
+ sort_keys=False,
+ allow_unicode=True,
+ default_flow_style=False,
+ width=10**9, # never fold: keep every scalar on one line so diffs stay readable
+ )
+ return f"---\n{fm}---\n\n{_rationale_body(rec)}\n"
+
+
+def _rationale_body(rec: dl.Record) -> str:
+ """The record body with its contiguous metadata block stripped — the prose, byte-for-byte.
+
+ Mirrors decisions_validate._rationale's contiguous-block rule, but preserves interior blank
+ lines and indentation (that function normalizes whitespace for comparison; here the bytes are
+ the payload).
+ """
+ lines = rec.body.splitlines()
+ i = 0
+ while i < len(lines) and not lines[i].strip():
+ i += 1
+ if i < len(lines) and lines[i].strip().startswith("`key:"):
+ j = i + 1
+ while j < len(lines) and lines[j].strip():
+ j += 1
+ lines = lines[:i] + lines[j:]
+ return "\n".join(lines).strip()
+
+
+def _preamble(path: Path, records: list[dl.Record]) -> str:
+ """Everything in `path` that is not inside a record — kept, never deleted."""
+ text = path.read_text(encoding="utf-8")
+ lines = text.splitlines()
+ drop: set[int] = set()
+ for r in records:
+ start = r.lineno - 1 # heading line, 1-indexed in Record
+ for n in range(start, start + 1 + len(r.body.splitlines())):
+ drop.add(n)
+ kept = [ln for n, ln in enumerate(lines) if n not in drop]
+ while kept and not kept[-1].strip():
+ kept.pop()
+ return "\n".join(kept)
+
+
+def plan() -> tuple[list[tuple[dl.Record, Path]], dict[Path, str]]:
+ """(record -> destination) pairs, and (source file -> preamble to keep)."""
+ sources = [p for p in dl.active_files() if p.exists()]
+ sources += sorted(p for p in dl.ARCHIVE_DIR.glob("*.md") if p.name not in dl._NON_DECISION_FILES)
+ moves: list[tuple[dl.Record, Path]] = []
+ preambles: dict[Path, str] = {}
+ for src in sources:
+ if dl.RECORDS_DIR in src.parents or dl.has_frontmatter(src.read_text(encoding="utf-8")):
+ continue # already migrated
+ recs = [r for r in dl.parse_file(src) if r.heading not in ("Index", "Active catalog", "Contents")]
+ keyed = [r for r in recs if r.key]
+ if len(keyed) != len(recs):
+ unkeyed = [r.heading for r in recs if not r.key]
+ raise SystemExit(f"{src}: {len(unkeyed)} record(s) without a key, cannot file by key: {unkeyed}")
+ moves += [(r, record_path(r)) for r in keyed]
+ preambles[src] = _preamble(src, recs)
+ return moves, preambles
+
+
+def main(argv=None) -> int:
+ ap = argparse.ArgumentParser()
+ ap.add_argument("--dry-run", action="store_true", help="report the plan, write nothing")
+ args = ap.parse_args(argv)
+
+ moves, preambles = plan()
+
+ dests = [d for _, d in moves]
+ if len(set(dests)) != len(dests):
+ dupes = sorted({str(d) for d in dests if dests.count(d) > 1})
+ raise SystemExit(f"destination collision — two records map to the same path: {dupes}")
+
+ print(f"{len(moves)} records -> {len({d.parent for d in dests})} directories")
+ print(
+ f"{len(preambles)} source files kept, stripped to narrative "
+ f"({sum(len(v.splitlines()) for v in preambles.values())} lines preserved)"
+ )
+ if args.dry_run:
+ for rec, dest in sorted(moves, key=lambda m: str(m[1]))[:5]:
+ print(f" {rec.key} -> {dest.relative_to(dl.REPO_ROOT)}")
+ print(" …")
+ return 0
+
+ for rec, dest in moves:
+ dest.parent.mkdir(parents=True, exist_ok=True)
+ dest.write_text(render_record(rec), encoding="utf-8")
+ for src, text in preambles.items():
+ src.write_text(text.rstrip("\n") + "\n", encoding="utf-8")
+ print("migration written")
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/scripts/tests/test_migration_equivalence.py b/scripts/tests/test_migration_equivalence.py
new file mode 100644
index 000000000..8d6c9ae95
--- /dev/null
+++ b/scripts/tests/test_migration_equivalence.py
@@ -0,0 +1,156 @@
+"""Equivalence harness for the #610 one-file-per-record migration.
+
+The migration is a MOVE: content relocates, nothing is rewritten. That makes correctness
+*provable* rather than reviewable — parse the corpus before, run the migration, parse it after,
+and assert the two Record sets are identical field-for-field with byte-identical rationale.
+
+These tests run the real migration against a COPY of the real corpus in tmp_path. They never touch
+the working tree.
+"""
+
+from __future__ import annotations
+
+import shutil
+from pathlib import Path
+
+import pytest
+
+import scripts.decisions_lib as dl
+import scripts.migrate_decisions_split as mig
+
+COMPARED_FIELDS = (
+ "key",
+ "status",
+ "since",
+ "supersedes",
+ "superseded_by",
+ "stale_after",
+ "rule",
+ "signals",
+ "mechanics",
+ "sources",
+)
+_SKIP = {"Index", "Active catalog", "Contents"}
+
+
+def _corpus_files() -> list[Path]:
+ files = [p for p in dl.active_files() if p.exists()]
+ files += [p for p in sorted(dl.ARCHIVE_DIR.glob("*.md")) if p.name not in dl._NON_DECISION_FILES]
+ return files
+
+
+@pytest.fixture(scope="module")
+def migrated(tmp_path_factory):
+ """Copy the real docs/ into tmp, point decisions_lib at it, run the migration."""
+ tmp = tmp_path_factory.mktemp("corpus")
+ shutil.copytree(dl.REPO_ROOT / "docs", tmp / "docs")
+
+ before = {}
+ for f in _corpus_files():
+ for r in dl.parse_file(f):
+ if r.key and r.heading not in _SKIP:
+ before[r.key] = (r, mig._rationale_body(r))
+
+ orig = (dl.REPO_ROOT, dl.DECISIONS_MD, dl.TOPIC_DIR, dl.ARCHIVE_DIR, dl.RECORDS_DIR)
+ dl.REPO_ROOT = tmp
+ dl.DECISIONS_MD = tmp / "docs" / "decisions.md"
+ dl.TOPIC_DIR = tmp / "docs" / "decisions"
+ dl.ARCHIVE_DIR = dl.TOPIC_DIR / "archive"
+ dl.RECORDS_DIR = dl.TOPIC_DIR / "records"
+ try:
+ mig.main([])
+ after_files = [p for p in dl.RECORDS_DIR.rglob("*.md")] + [p for p in dl.ARCHIVE_DIR.rglob("*.md")]
+ after = {}
+ for f in after_files:
+ for r in dl.parse_file(f):
+ if r.key:
+ after[r.key] = (r, r.body)
+ yield before, after, tmp
+ finally:
+ dl.REPO_ROOT, dl.DECISIONS_MD, dl.TOPIC_DIR, dl.ARCHIVE_DIR, dl.RECORDS_DIR = orig
+
+
+def test_every_record_survives_with_the_same_key(migrated):
+ before, after, _ = migrated
+ assert before, "harness parsed nothing before the migration — would pass vacuously"
+ assert set(before) == set(after), (
+ f"lost: {sorted(set(before) - set(after))[:5]} / gained: {sorted(set(after) - set(before))[:5]}"
+ )
+
+
+def test_record_count_is_the_expected_166(migrated):
+ before, after, _ = migrated
+ assert len(before) == len(after) == 166
+
+
+def test_every_metadata_field_round_trips(migrated):
+ before, after, _ = migrated
+ diffs = []
+ for key, (b, _) in before.items():
+ a, _ = after[key]
+ for f in COMPARED_FIELDS:
+ if getattr(b, f) != getattr(a, f):
+ diffs.append(f"{key}.{f}: {getattr(b, f)!r} -> {getattr(a, f)!r}")
+ assert not diffs, "\n".join(diffs[:10])
+
+
+def test_title_round_trips_from_the_heading(migrated):
+ before, after, _ = migrated
+ diffs = [
+ f"{k}: {b.heading!r} -> {after[k][0].heading!r}"
+ for k, (b, _) in before.items()
+ if b.heading != after[k][0].heading
+ ]
+ assert not diffs, "\n".join(diffs[:10])
+
+
+def test_rationale_bodies_are_byte_identical(migrated):
+ before, after, _ = migrated
+ diffs = [k for k, (_, body) in before.items() if body != after[k][1]]
+ assert not diffs, f"{len(diffs)} rationale bodies changed, e.g. {diffs[:5]}"
+
+
+def test_path_matches_key(migrated):
+ _, after, _tmp = migrated
+ bad = []
+ for key, (rec, _) in after.items():
+ area, _, topic = key.partition(".")
+ if rec.source.parent.name != area or rec.source.stem != topic:
+ bad.append(f"{key} -> {rec.source}")
+ assert not bad, "\n".join(bad[:10])
+
+
+def test_archived_records_land_in_the_archive_wing(migrated):
+ _, after, _tmp = migrated
+ misfiled = [
+ key
+ for key, (rec, _) in after.items()
+ if (rec.status in ("superseded", "retired")) != ("archive" in rec.source.parts)
+ ]
+ assert not misfiled, misfiled
+
+
+def test_source_files_are_kept_and_retain_their_narrative(migrated):
+ """The ~500 lines of preamble are the only copy of that prose — they must survive."""
+ _, _, tmp = migrated
+ kept = tmp / "docs" / "decisions.md"
+ assert kept.exists(), "decisions.md was deleted — its lifecycle schema is the only copy"
+ text = kept.read_text()
+ assert "Lifecycle, not append-only" in text
+ # NB: do NOT string-match on "## " here — the schema header quotes an illustrative
+ # "## 2026-07-17 — …" example in prose. Whether real records remain is a PARSING question,
+ # covered by test_no_records_remain_in_the_legacy_files.
+ wp = tmp / "docs" / "decisions" / "workflow-process.md"
+ assert wp.exists() and "extracted from" in wp.read_text()
+
+
+def test_no_records_remain_in_the_legacy_files(migrated):
+ _, _, tmp = migrated
+ leftovers = {}
+ for f in [tmp / "docs" / "decisions.md", *sorted((tmp / "docs" / "decisions").glob("*.md"))]:
+ if f.name in dl._NON_DECISION_FILES:
+ continue
+ keyed = [r.key for r in dl.parse_file(f) if r.key]
+ if keyed:
+ leftovers[f.name] = keyed
+ assert not leftovers, leftovers