scripts.migrate_legacy_story_ids

Reference › scripts › migrate_legacy_story_ids

scripts/migrate_legacy_story_ids.py rewrites legacy requirement ids of the form REQ-NNN-USn-MMM to the canonical REQ-NNN-US-NNN-MMM, together with the matching spec headings. A human runs it against the repository; it has no internal imports.

Options: --root (default the repository root), --apply (write changes; the default is a dry run) and --report-collisions. canonical_story pads the story number to three digits, migrate_text rewrites every legacy id and heading in a text, iter_candidate_files lists the in-scope files in sorted order (with is_excluded and _matches applying the exclusion globs), and run returns the files changed, the substitutions and the skipped files. The safety check is find_collisions: it maps each canonical target to the distinct legacy spellings that reach it. If any collision is not in KNOWN_BENIGN_COLLISIONS, a real migration refuses to run, because it would merge two distinct requirement ids into one; the message directs the user to --report-collisions. That option prints every collision tagged known-benign or unrecognised and exits 1 if any is unrecognised.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Constants

Classes

ManglingError

Bases: RuntimeError · Line 200

Raised when a migration would change anything other than a story id.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/migrate_legacy_story_ids.py

#!/usr/bin/env python3
"""Migrate legacy story ids (``REQ-NNN-USn-MMM``) to canonical ``REQ-NNN-US-NNN-MMM``.

Canonical grammar (``docs/req-id-grammar.md``): the story segment is ``US-``
followed by exactly three digits. Three LEGACY shapes exist in the corpus and
all three are rewritten:

    REQ-002-USn-001    ->  REQ-002-US-001-001     (n = 1 digit,  e.g. US1)
    REQ-001-USnn-004   ->  REQ-001-US-010-004     (nn = 2 digits, e.g. US10)
    REQ-004-USnnn-001  ->  REQ-004-US-001-001     (nnn = 3 digits, dash missing)

NOTE: the left column deliberately uses the ``USn`` PLACEHOLDER spelling, not
a literal legacy id. Written literally, this migrator rewrites its own
documentation on the next run and the examples degrade into ``X -> X``
identities that teach nothing. That happened once, on 2026-08-19, and was
caught by reading the diff rather than by any check — the safety invariants
cannot help here, because an illustrative id is structurally indistinguishable
from a real one. ``docs/req-id-grammar.md`` treats these placeholder tokens
the same way, and ``check_req_format.py``'s ``PLACEHOLDER`` pattern exempts
them for exactly this reason.

Measured 2026-08-19 before migration: 61,700 one-digit, 7,992 two-digit and
13,444 three-digit occurrences (83,136 total) against 1,925 already-canonical.

WHY THIS IS NOT A ONE-LINE SED
------------------------------
A naive ``s/US([0-9])-/US-00\\1-/`` matches only the 1-digit shape. It leaves
21,436 two- and three-digit ids behind while *reporting success*, and the
survivors are the ones a human is least likely to notice. Every shape is
handled here by a single regex with an explicit ``{1,3}`` width and a
normalise-then-pad step, so a legacy id cannot be silently skipped.

Equally important, ``US-001`` (already canonical) must NOT match. The pattern
requires a digit immediately after ``US``, which a canonical id never has.

COLLISIONS
----------
Two legacy ids can normalise onto one canonical id (``US1`` and ``US001`` both
-> ``US-001``). Measured across the whole corpus: 1,851 distinct legacy ids
collapse to 1,849 canonical, i.e. exactly 2 collisions, both verified benign —
``REQ-007-US-001-001`` and ``REQ-999-US-001-001`` occur only as synthetic strings
inside parser tests, never as real requirement definitions. ``--report-collisions``
recomputes this rather than trusting the note: a NEW collision between two real
requirements would silently merge them, so the migration refuses to run if one
appears that is not in ``KNOWN_BENIGN_COLLISIONS``.

SCOPE
-----
Derived artefacts (``plans/ast-*.json``, ``.cache/``) are NOT rewritten — they
are regenerated from source afterwards, so editing them would be both wasted
work and a second source of truth. ``.git``/``.venv``/``worktrees``/
``node_modules`` are never touched.
"""

from __future__ import annotations

import argparse
import re
import sys
from collections import defaultdict
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent

# A legacy story segment: "US" followed immediately by 1-3 digits, then "-".
# The digit immediately after "US" is what distinguishes legacy from canonical
# ("US-001" has a dash there and therefore cannot match).
LEGACY_ID_RE = re.compile(r"\bREQ-(\d{3})-US(\d{1,3})-(\d+)\b")

# The same story segment in a spec heading ("## US1: ..." / "### US10").
# MULTILINE so it applies across a whole document in one pass; a canonical
# "## US-004" cannot match because a digit must follow "US" directly.
LEGACY_HEADING_RE = re.compile(r"^(#{2,4}\s+)US(\d{1,3})\b", re.MULTILINE)

# Directories whose contents are derived, foreign, or per-run state; never
# rewritten. ``pipeline_checkpoints`` is gitignored runtime state (0 tracked
# files, 331 live dirs measured 2026-08-19): its ids are a snapshot of a past
# run, so rewriting them edits history while changing nothing a reader
# depends on.
EXCLUDED_DIR_PARTS: frozenset[str] = frozenset(
    {
        ".git",
        ".venv",
        "worktrees",
        "node_modules",
        "__pycache__",
        ".cache",
        ".mypy_cache",
        "pipeline_checkpoints",
        # Historical run records. Same reasoning as pipeline_checkpoints: a log
        # states what a past run DID emit, so rewriting it makes the record
        # lie about the run it describes.
        "logs",
        # Runtime scratch + sibling worktree checkouts. `.claude/` (dotted) is
        # machine state; `claude/` (undotted) is production source and IS in
        # scope.
        ".claude",
    }
)

# Derived AST snapshots — regenerated from source after the migration — plus
# the two test files that must keep genuine legacy examples forever in order to
# prove legacy is still recognised.
#
# A DETECTOR's test needs a positive fixture, and migrating that fixture does
# not make the corpus cleaner — it makes the detector untested. The first
# migration run proved this the hard way: it rewrote the legacy fixture in
# test_check_req_format.py, leaving ``test_legacy_form_counted`` and
# ``test_canonical_form_not_counted`` holding IDENTICAL input while asserting
# opposite results. A contradiction, reached silently, by a codemod whose
# invariants were all satisfied — because "did I destroy a negative fixture?"
# is a question about MEANING, which no byte-level invariant can ask.
EXCLUDED_GLOBS: tuple[str, ...] = (
    "plans/ast-*.json",
    "tests/scripts/test_migrate_legacy_story_ids.py",
    "tests/scripts/test_migrate_legacy_story_ids_coverage.py",
    "tests/scripts/test_check_req_format.py",
    # ── added 2026-08-23, each measured, each enumerated ──────────────────
    #
    # reports/* — the corpus guard's OWN OUTPUT, and the reason it could
    # never go green again once it had failed once. The assertion in
    # tests/scripts/test_migrate_legacy_story_ids.py names up to 25
    # offenders IN ITS MESSAGE; run_tests.py records that message into
    # reports/run-tests-latest.json; the next scan reads those quoted ids
    # back as fresh offenders. Measured: all 58 occurrences in that file sit
    # inside the recorded failure text, and nowhere else. A guard whose
    # failure manufactures its own next failure is not a ratchet, it is a
    # latch. reports/ is test OUTPUT — nothing in it is source to migrate.
    "reports/*",
    #
    # plans/analysis/*.json — a DERIVED register. plans/analysis/README.md
    # states the pipeline: journal.jsonl (per-round swarm output) ->
    # build_register.py -> register.json -> render_report.py. Its legacy ids
    # are QUOTED EVIDENCE of what an agent reported seeing in the sources it
    # read. Rewriting them would be wrong twice: build_register.py reverts
    # it on the next regeneration, and a rewritten quote claims the agent
    # saw something it did not. Same class as plans/ast-*.json above.
    "plans/analysis/*.json",
    #
    # plugins/orchestrata/** — the GENERATED mirror of claude/ and .claude/,
    # written by claude/scripts/gen_plugin.py. `.claude` is already excluded
    # via EXCLUDED_DIR_PARTS, so before this the scan skipped the SOURCE
    # hook and reported only its byte-identical generated copy — the one
    # place a fix is forbidden, because gen_plugin.py overwrites it. The two
    # ids involved are illustrative examples inside comments -- a sample
    # `grep -rn` invocation quoting a legacy-shaped id -- the same kind of
    # legitimate legacy-id quotation the two test files above are excluded
    # for. (Written masked here on purpose: quoting one literally would make
    # THIS file an offender, which is how the first draft of this comment
    # tripped the very guard it documents.)
    #
    # Scoped to plugins/orchestrata/ and NOT plugins/ or a `plugins`
    # EXCLUDED_DIR_PARTS entry: either of those also swallows
    # tests/plugins/ (hand-written suite plugins) and opencode/plugins/,
    # blinding the guard to two real source trees.
    "plugins/orchestrata/**",
)

# Extensions worth rewriting. Binary and lockfiles are excluded by omission.
# ``.mmd`` (Mermaid) earns its place the hard way: the first sweep omitted it
# and left 94 legacy ids across 5 architecture diagrams in plans/contracts/ --
# real source that a suffix allowlist silently skipped while reporting
# success. When adding a scope rule, enumerate what it EXCLUDES, not only what
# it includes.
INCLUDED_SUFFIXES: frozenset[str] = frozenset(
    {
        ".py",
        ".md",
        ".pl",
        ".json",
        ".yaml",
        ".yml",
        ".toml",
        ".txt",
        ".ndjson",
        ".mmd",
        ".sh",
        ".cfg",
        ".ini",
    }
)

# (spec, canonical_story, seq) tuples where two legacy spellings legitimately
# collapse onto one canonical id. Verified 2026-08-19: both are synthetic
# strings inside parser tests, not competing requirement definitions.
KNOWN_BENIGN_COLLISIONS: frozenset[tuple[str, str, str]] = frozenset(
    {("007", "001", "001"), ("999", "001", "001")}
)


def canonical_story(raw: str) -> str:
    """``"1"``/``"01"``/``"001"`` -> ``"001"``; ``"10"`` -> ``"010"``.

    Strips leading zeros then pads to three, so every legacy width lands on
    one canonical spelling.
    """
    return str(int(raw)).zfill(3)


class ManglingError(RuntimeError):
    """Raised when a migration would change anything other than a story id.

    This is a stop-the-world condition: it means the rewrite touched bytes it
    had no business touching, and the file must NOT be written.
    """


# Matches a story id in EITHER spelling (legacy ``US1``/``US10``/``US001`` or
# canonical ``US-001``), used only to blank ids out for the safety invariant.
_ANY_ID_RE = re.compile(r"\bREQ-\d{3}-US-?\d{1,3}-\d+\b")
_ANY_HEADING_RE = re.compile(r"^(#{2,4}\s+)US-?\d{1,3}\b", re.MULTILINE)

_ID_SENTINEL = "\x00ID\x00"
_HEADING_SENTINEL = "\x00US\x00"


def skeleton(text: str) -> str:
    """``text`` with every story id and story heading blanked to a sentinel.

    Two texts with equal skeletons differ ONLY in story ids. This is the
    safety invariant: a correct migration must leave the skeleton untouched,
    which is a far stronger guarantee than any list of cases I thought to
    test -- it catches a mangled neighbouring token, a swallowed character,
    a dropped line, or a mauled encoding, whether or not it was anticipated.
    """
    blanked = _ANY_ID_RE.sub(_ID_SENTINEL, text)
    return _ANY_HEADING_RE.sub(lambda m: m.group(1) + _HEADING_SENTINEL, blanked)


def migrate_text(text: str) -> tuple[str, int]:
    """Rewrite every legacy id and spec heading in ``text``.

    Returns the new text and the number of substitutions made.

    Raises:
        ManglingError: the rewrite altered something other than a story id.
    """
    count = 0

    def _id(match: re.Match[str]) -> str:
        nonlocal count
        count += 1
        spec, story, seq = match.groups()
        return f"REQ-{spec}-US-{canonical_story(story)}-{seq}"

    result = LEGACY_ID_RE.sub(_id, text)

    def _heading(match: re.Match[str]) -> str:
        nonlocal count
        count += 1
        return f"{match.group(1)}US-{canonical_story(match.group(2))}"

    # Applied directly: LEGACY_HEADING_RE requires a DIGIT immediately after
    # "US", which a canonical "US-004" heading never has, so canonical
    # headings cannot match and need no second filter. Routing this through
    # _ANY_HEADING_RE instead handed _heading a one-group match and raised
    # IndexError on group(2) -- caught by the heading sample before any file
    # was written.
    result = LEGACY_HEADING_RE.sub(_heading, result)

    # Invariant 1: nothing outside a story id may change.
    if skeleton(result) != skeleton(text):
        raise ManglingError(
            "migration altered content outside a story id -- refusing to write"
        )

    # Invariant 2: the rewrite only ever LENGTHENS (US1->US-001 adds 3 chars,
    # US10->US-010 adds 2, US001->US-001 adds 1). A shorter result means
    # something was deleted.
    if len(result) < len(text):
        raise ManglingError(
            f"migration shortened content by {len(text) - len(result)} char(s) "
            "-- refusing to write"
        )

    # Invariant 3: line count is preserved exactly.
    if result.count("\n") != text.count("\n"):
        raise ManglingError("migration changed the line count -- refusing to write")

    return result, count


def is_excluded(path: Path, root: Path | None = None) -> bool:
    """True when ``path`` is derived, foreign, or a non-text artefact.

    ``root`` anchors the derived-artefact globs. It defaults to the repo so
    callers scanning this tree need not pass it, but it MUST be honoured for
    any other tree: resolving against a hardcoded repo root raises ValueError
    the moment ``path`` lives elsewhere, which is precisely what a temp-tree
    test exposed before this migration ever touched real files.
    """
    if any(part in EXCLUDED_DIR_PARTS for part in path.parts):
        return True
    if path.suffix.lower() not in INCLUDED_SUFFIXES:
        return True
    base = root if root is not None else REPO_ROOT
    try:
        rel = path.relative_to(base) if path.is_absolute() else path
    except ValueError:
        rel = Path(path.name) if path.parent == path.parent.parent else path
    return any(_matches(rel, pattern) for pattern in EXCLUDED_GLOBS)


def _matches(rel: Path, pattern: str) -> bool:
    """One glob test, using the matcher whose semantics the pattern implies.

    ``Path.match`` does NOT treat ``**`` as recursive and matches from the
    RIGHT, so a pattern that merely LOOKS recursive is silently inert -- the
    failure class this module's own suffix-allowlist comment warns about.
    Measured 2026-08-23 on Python 3.14.6, against the real candidates:

        plugins/orchestrata/**  .match      -> matches NOTHING
        plugins/orchestrata/**  .full_match -> matches the mirror, recursively
        plugins/**              .match      -> matches tests/plugins/... and
                                               opencode/plugins/... but NOT
                                               plugins/orchestrata/...

    That third line is why this is a function and not a one-character edit:
    the naive recursive-looking pattern excludes precisely the two trees that
    must stay in scope, and none of the one that should leave it.

    ``full_match`` is used ONLY for patterns containing ``**``. Every pattern
    without it keeps ``match``'s existing right-anchored behaviour, so this is
    additive -- no pattern already in EXCLUDED_GLOBS changes meaning.
    """
    if "**" in pattern:
        return rel.full_match(pattern)
    return rel.match(pattern)


def iter_candidate_files(root: Path) -> list[Path]:
    """Every in-scope file under ``root``, sorted for deterministic output."""
    return sorted(
        p for p in root.rglob("*") if p.is_file() and not is_excluded(p, root)
    )


def find_collisions(root: Path) -> dict[tuple[str, str, str], set[str]]:
    """Map each canonical target to the distinct legacy ids that reach it.

    Only entries with more than one source are returned — those are the ids
    where migration MERGES two spellings and therefore needs a human verdict.
    """
    targets: dict[tuple[str, str, str], set[str]] = defaultdict(set)
    for path in iter_candidate_files(root):
        try:
            text = path.read_text(encoding="utf-8")
        except OSError, UnicodeDecodeError:
            continue
        for match in LEGACY_ID_RE.finditer(text):
            spec, story, seq = match.groups()
            targets[(spec, canonical_story(story), seq)].add(match.group(0))
    return {k: v for k, v in targets.items() if len(v) > 1}


def run(root: Path, apply: bool) -> tuple[int, int, int]:
    """Migrate under ``root``. Returns (files_changed, substitutions, skipped)."""
    files_changed = substitutions = skipped = 0
    for path in iter_candidate_files(root):
        try:
            original = path.read_text(encoding="utf-8")
        except OSError, UnicodeDecodeError:
            skipped += 1
            continue
        migrated, count = migrate_text(original)
        if count == 0 or migrated == original:
            continue
        files_changed += 1
        substitutions += count
        if apply:
            try:
                path.write_text(migrated, encoding="utf-8")
            except OSError:
                skipped += 1
    return files_changed, substitutions, skipped


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="migrate_legacy_story_ids",
        description="Rewrite legacy REQ-NNN-USn-MMM ids to canonical REQ-NNN-US-NNN-MMM.",
    )
    parser.add_argument("--root", default=str(REPO_ROOT), help="Tree to migrate.")
    parser.add_argument(
        "--apply", action="store_true", help="Write changes (default: dry run)."
    )
    parser.add_argument(
        "--report-collisions",
        action="store_true",
        help="List canonical ids reachable from more than one legacy spelling.",
    )
    args = parser.parse_args(argv)
    root = Path(args.root).resolve()

    collisions = find_collisions(root)
    unknown = {k: v for k, v in collisions.items() if k not in KNOWN_BENIGN_COLLISIONS}

    if args.report_collisions:
        print(f"collisions: {len(collisions)} ({len(unknown)} unrecognised)")
        for key, sources in sorted(collisions.items()):
            tag = "KNOWN-BENIGN" if key in KNOWN_BENIGN_COLLISIONS else "UNRECOGNISED"
            print(f"  [{tag}] REQ-{key[0]}-US-{key[1]}-{key[2]} <- {sorted(sources)}")
        return 1 if unknown else 0

    if unknown:
        print(
            f"REFUSING: {len(unknown)} unrecognised collision(s) would MERGE two "
            f"distinct requirement ids into one. Re-run with --report-collisions, "
            f"verify each is benign, then add it to KNOWN_BENIGN_COLLISIONS.",
            file=sys.stderr,
        )
        for key, sources in sorted(unknown.items()):
            print(
                f"  REQ-{key[0]}-US-{key[1]}-{key[2]} <- {sorted(sources)}",
                file=sys.stderr,
            )
        return 2

    files_changed, substitutions, skipped = run(root, apply=args.apply)
    mode = "APPLIED" if args.apply else "DRY RUN"
    print(
        f"{mode}: {substitutions} substitution(s) across {files_changed} file(s); "
        f"{skipped} skipped; {len(collisions)} known-benign collision(s)."
    )
    if not args.apply:
        print("Re-run with --apply to write.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())