fix: report migrations chained behind the deployed head

README.md

@@ -21,7 +21,7 @@ Instead of hardcoding `down_revision`, this library determines the migration cha
 This means:
 - New migrations never conflict with each other
 - The chain is always linear, regardless of branch merge order
-- Existing migrations with hardcoded `down_revision` continue to work
+- Existing migrations with hardcoded `down_revision` continue to work, as long as they are not chained behind a revision that has already been applied somewhere (see [Hardcoded `down_revision` on a deployed database](#hardcoded-down_revision-on-a-deployed-database))
 
 ## Installation
 
@@ -91,6 +91,49 @@ The library handles three types of migrations:
 
 Classification reads the `revision` and `down_revision` attributes from each migration module (the same values Alembic loads), so any Alembic `file_template` and any `rev_id` format work.
 
+## Hardcoded `down_revision` on a deployed database
+
+A hybrid is placed immediately after the revision it points at, so any dynamic migration added between the two is re-parented onto the hybrid. That is what keeps the chain linear when two branches fork from the same head, and it is safe while none of those migrations has run yet.
+
+It stops being safe once one of them has been applied. Say `bbbb -> cccc -> dddd` are already deployed, and a new migration hardcodes `down_revision = "bbbb"`:
+
+```
+chain before:  aaaa -> bbbb -> cccc -> dddd          alembic_version = dddd
+chain after:   aaaa -> bbbb -> eeee -> cccc -> dddd  alembic_version = dddd
+```
+
+`eeee` now sits behind the recorded head. `alembic upgrade head` walks down from `dddd`, finds every revision already applied, and does nothing. The migration never runs, and nothing reports an error.
+
+**The rule: only hardcode `down_revision` onto the current head.** Anything older is spliced into history the database has already walked past. Prefer `get_down_revision(revision)`, which cannot pick a stale parent.
+
+To check an existing tree:
+
+```bash
+alembic-git-revisions --check /path/to/versions
+```
+
+This lists every hybrid chained ahead of earlier revisions and **exits 0**, because git history alone cannot prove a finding is a real problem. A hybrid added by a branch that forked from the same head produces exactly the same shape, and that case is benign. Treat the output as something to look at, not as a failure.
+
+Only the revisions a database has actually applied separate the two. Pass them in to narrow the report to migrations that provably cannot run, which **exits non-zero**:
+
+```bash
+psql -Atc 'select version_num from alembic_version' \
+  | alembic-git-revisions --check --applied - /path/to/versions
+```
+
+That is the form worth gating CI on. Note `alembic_version` records only the current head, so for a full picture supply every revision reachable from it. The same distinction is available from Python:
+
+```python
+from alembic_git_revisions import find_displaced_revisions
+
+# advisory: includes the benign fork-from-the-same-head case
+candidates = find_displaced_revisions(versions_dir)
+
+# confirmed: only what cannot run on this database
+for found in find_displaced_revisions(versions_dir, applied=applied):
+    print(f"{found.hybrid} will never run on this database")
+```
+
 ## API
 
 ### `get_down_revision(revision, versions_dir=None)`
@@ -127,6 +170,10 @@ A frozen dataclass describing one parsed migration:
 
 `git_sequence` is a position within one particular parse, not a stable property of the file. Files absent from git history all share the same end-of-list sentinel and are separated only by `filename`, which is why `parse_versions_dir` sorts on both.
 
+### `find_displaced_revisions(versions_dir, applied=None)`
+
+Returns a list of `DisplacedRevision(hybrid, target, displaced)`, one per hybrid that is chained ahead of revisions added before it. Without `applied` the result is advisory and includes the benign fork-from-the-same-head case. Pass `applied`, the revisions a database has actually run, to keep only the hybrids that can never execute on it. See [Hardcoded `down_revision` on a deployed database](#hardcoded-down_revision-on-a-deployed-database).
+
 ### `CHAIN_FILENAME`
 
 Name of the generated chain file, `revision_chain.json`. Use it instead of hardcoding the string when locating or cleaning up the generated artifact.

alembic_git_revisions/__init__.py

@@ -17,18 +17,25 @@
 
 from __future__ import annotations
 
+import argparse
 import pathlib
 import sys
 
 from alembic_git_revisions._chain import (
     CHAIN_FILENAME as CHAIN_FILENAME,
 )
+from alembic_git_revisions._chain import (
+    DisplacedRevision as DisplacedRevision,
+)
 from alembic_git_revisions._chain import (
     MigrationFile as MigrationFile,
 )
 from alembic_git_revisions._chain import (
     build_chain as build_chain,
 )
+from alembic_git_revisions._chain import (
+    find_displaced_revisions as find_displaced_revisions,
+)
 from alembic_git_revisions._chain import (
     generate_chain_file as generate_chain_file,
 )
@@ -40,13 +47,123 @@
 )
 
 
-def _cli() -> None:
-    """CLI entry point: generate revision_chain.json."""
-    if len(sys.argv) != 2:  # noqa: PLR2004
+def _read_applied(source: str) -> set[str]:
+    """Read applied revision ids, one per line, from a file or stdin (``-``).
+
+    Blank lines and ``#`` comments are ignored, so the output of a query
+    against ``alembic_version`` can be piped in directly.
+    """
+    text = (
+        sys.stdin.read()
+        if source == "-"
+        else pathlib.Path(source).read_text(encoding="utf-8")
+    )
+    return {
+        stripped
+        for line in text.splitlines()
+        if (stripped := line.strip()) and not stripped.startswith("#")
+    }
+
+
+def _report_candidates(displaced: list[DisplacedRevision]) -> int:
+    """Report hybrids that *may* be unreachable. Always exit code 0.
+
+    Without the set of applied revisions nothing here is known to be wrong:
+    two branches forking from the same head produce this shape legitimately.
+    Reporting a non-zero status would fail that ordinary workflow.
+    """
+    if not displaced:
+        print("No hybrid migrations are chained ahead of earlier revisions.")  # noqa: T201
+        return 0
+
+    for revision in displaced:
+        following = ", ".join(revision.displaced)
         print(  # noqa: T201
-            f"Usage: {sys.argv[0]} <versions-directory>",
+            f"{revision.hybrid}: hardcodes down_revision={revision.target!r}; "
+            f"{len(revision.displaced)} later revision(s) now chain after it: "
+            f"{following}\n"
+            f"  If any of those is already applied on a database, "
+            f"{revision.hybrid} sits behind that database's head and will not "
+            f"run there. Git history alone cannot tell: two branches forking "
+            f"from the same head produce this shape legitimately. Pass "
+            f"--applied to decide.",
+        )
+    return 0
+
+
+def _report_confirmed(displaced: list[DisplacedRevision], applied: set[str]) -> int:
+    """Report hybrids that cannot run on the described database.
+
+    Exits non-zero, because every finding here is backed by a revision the
+    database has actually applied.
+    """
+    if not displaced:
+        print("No migration is chained behind an applied revision.")  # noqa: T201
+        return 0
+
+    for revision in displaced:
+        blocking = ", ".join(r for r in revision.displaced if r in applied)
+        print(  # noqa: T201
+            f"{revision.hybrid}: hardcodes down_revision={revision.target!r}, "
+            f"but {blocking} is already applied and now chains after it.\n"
+            f"  {revision.hybrid} sits behind this database's head, so "
+            f"'alembic upgrade head' will never run it.",
             file=sys.stderr,
         )
-        sys.exit(1)
+    return 1
+
 
-    generate_chain_file(pathlib.Path(sys.argv[1]))
+def _cli() -> None:
+    """CLI entry point: generate revision_chain.json, or check the chain."""
+    parser = argparse.ArgumentParser(
+        prog="alembic-git-revisions",
+        description=(
+            "Generate revision_chain.json from git history, or with --check "
+            "report hybrid migrations chained ahead of earlier revisions."
+        ),
+    )
+    parser.add_argument(
+        "versions_dir",
+        metavar="versions-directory",
+        type=pathlib.Path,
+        help="Alembic versions directory.",
+    )
+    parser.add_argument(
+        "--check",
+        action="store_true",
+        help=(
+            "Report hybrid migrations chained ahead of revisions added "
+            "before them, instead of generating the chain file. Exits 0: "
+            "git history alone cannot prove a finding is a real problem."
+        ),
+    )
+    parser.add_argument(
+        "--applied",
+        metavar="FILE",
+        help=(
+            "With --check, a file of applied revision ids (one per line, "
+            "'-' for stdin) as recorded by the target database. Narrows the "
+            "report to migrations that provably cannot run, and exits "
+            "non-zero if there are any."
+        ),
+    )
+    args = parser.parse_args()
+
+    if args.applied is not None and not args.check:
+        parser.error("--applied requires --check")
+
+    # Both paths fail the same way when git is missing or shallow, and an
+    # unreadable --applied file is an ordinary user error. Report all of
+    # them as a message rather than a traceback.
+    try:
+        if args.check:
+            applied = _read_applied(args.applied) if args.applied is not None else None
+            displaced = find_displaced_revisions(args.versions_dir, applied=applied)
+            if applied is None:
+                sys.exit(_report_candidates(displaced))
+            sys.exit(_report_confirmed(displaced, applied))
+
+        generate_chain_file(args.versions_dir)
+    except (RuntimeError, OSError) as exc:
+        print(f"{parser.prog}: error: {exc}", file=sys.stderr)  # noqa: T201
+        sys.exit(1)

alembic_git_revisions/_chain.py

@@ -9,6 +9,12 @@
 import re
 import subprocess
 
+# Imported at runtime rather than under TYPE_CHECKING: this annotates public
+# API, and ``from __future__ import annotations`` turns annotations into
+# strings, so the name has to stay in module globals for
+# ``typing.get_type_hints()`` to resolve it.
+from collections.abc import Collection  # noqa: TC003
+
 _REVISION_FROM_FILENAME_RE = re.compile(r"^([a-f0-9]+)_")
 
 CHAIN_FILENAME = "revision_chain.json"
@@ -531,3 +537,80 @@ def generate_chain_file(versions_dir: pathlib.Path) -> None:
         json.dump(chain, f, indent=2, sort_keys=True)
         f.write("\n")
     print(f"Generated {chain_file} with {len(chain)} revisions")  # noqa: T201
+
+
+@dataclasses.dataclass(frozen=True)
+class DisplacedRevision:
+    """A hybrid migration chained ahead of revisions that predate it.
+
+    ``hybrid`` hardcodes ``target`` as its ``down_revision``.  Because a
+    hybrid is placed immediately after its target, every revision in
+    ``displaced`` -- added to git after the target but before the hybrid --
+    is re-parented onto the hybrid instead of staying where it was.
+
+    That is intended while none of them has run anywhere: it is what keeps
+    the chain linear when two branches fork from the same head.  It is
+    destructive once any of them has been applied.  A database whose
+    ``alembic_version`` records a displaced revision has already walked past
+    the point where the hybrid now sits, so ``alembic upgrade head`` finds
+    nothing to do and the hybrid never runs.
+
+    Git history cannot separate the two cases; only the set of applied
+    revisions can.  Pass one to :func:`find_displaced_revisions` to narrow
+    candidates down to the hybrids that are genuinely unreachable.
+    """
+
+    hybrid: str
+    target: str
+    displaced: tuple[str, ...]
+
+
+def _displaced_revisions(files: list[MigrationFile]) -> list[DisplacedRevision]:
+    """Find every hybrid that displaces revisions added before it."""
+    dynamic_revisions = {f.revision for f in files if f.is_dynamic}
+    sequence = {f.revision: f.git_sequence for f in files}
+
+    found: list[DisplacedRevision] = []
+    for hybrid in files:
+        if hybrid.is_dynamic:
+            continue
+        for target in hybrid.static_down_revisions:
+            if target not in dynamic_revisions:
+                continue
+            target_sequence = sequence[target]
+            displaced = tuple(
+                f.revision
+                for f in files
+                if f.is_dynamic
+                and target_sequence < f.git_sequence < hybrid.git_sequence
+            )
+            if displaced:
+                found.append(
+                    DisplacedRevision(hybrid.revision, target, displaced),
+                )
+    return found
+
+
+def find_displaced_revisions(
+    versions_dir: pathlib.Path,
+    applied: Collection[str] | None = None,
+) -> list[DisplacedRevision]:
+    """Report hybrids chained ahead of revisions that were added before them.
+
+    Without *applied* the result is advisory: it lists every hybrid whose
+    target is not the newest revision preceding it, which includes the benign
+    case of two branches forking from the same head.
+
+    Pass *applied* -- the revisions a database has actually run, e.g. read
+    from ``alembic_version`` or an equivalent record -- to keep only the
+    findings that matter.  A hybrid displacing an already-applied revision
+    sits behind that database's head and will never be executed.
+    """
+    found = _displaced_revisions(parse_versions_dir(versions_dir))
+    if applied is None:
+        return found
+    return [
+        revision
+        for revision in found
+        if any(displaced in applied for displaced in revision.displaced)
+    ]