migration_preflight.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/scripts/migration_preflight.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/unit/test_migration_preflight.py"
"""
M3.1 monorepo migration pre-flight artifact generator.
 
Purpose
-------
Capture the non-destructive baselines needed before the future M3.2
``git filter-repo`` migration into the Option_C monorepo.
 
Responsibilities
----------------
- Write branch/worktree inventory artifacts.
- Capture ``oct health --json`` output or a structured skip stub.
- Record the pytest collection count used for later parity checks.
- Run a filter-repo rehearsal only inside a throwaway temp clone.
- Write a summary artifact that separates automated and manual steps.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - CLI lifecycle and artifact writes
    L3 - subprocess and tool-availability decisions
    L4 - row-level inventory details
 
Contracts
---------
- No live branch or repository history is rewritten.
- Missing optional tools write SKIPPED artifacts and return success.
- Subprocesses run with argument lists and never through ``shell=True``.
- The filter-repo command is never run against the source repository root.
 
Dependencies
------------
- git (not imported; CLI run as a subprocess for the branch and worktree inventory, the temp clone and the optional git filter-repo rehearsal)
- pytest (not imported; run as python -m pytest --collect-only for the collection-count baseline)
- oct (not imported; optional CLI run as oct health --json for the health baseline)
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
import argparse
from dataclasses import dataclass
from datetime import datetime, timezone
import json
from pathlib import Path
import re
import shutil
import subprocess
import sys
import tempfile
from typing import Sequence
 
from oc_diagnostics import _dbg
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
DEFAULT_OUT_DIR = ".migration"
DEFAULT_TIMEOUT_SECONDS = 300
COUNT_RE = re.compile(r"(\d+)\s+(?:tests?|items?)\s+collected", re.IGNORECASE)
PER_FILE_COUNT_RE = re.compile(r"^[^:\n]+:\s*(\d+)\s*$", re.MULTILINE)
 
 
@dataclass(frozen=True)
class Stamp:
    """Shared artifact timestamp for one invocation."""
 
    date: str
    full: str
 
 
@dataclass(frozen=True)
class BranchRow:
    """One local branch inventory row."""
 
    name: str
    sha: str
    date: str
    subject: str
    ahead_behind: str
    merged: str
 
 
@dataclass(frozen=True)
class WorktreeRow:
    """One git worktree inventory row."""
 
    path: str
    branch: str
    sha: str
    date: str
    subject: str
    ahead_behind: str
    merged: str
 
 
def _iso_now() -> Stamp:
    now = datetime.now(timezone.utc).replace(microsecond=0)
    return Stamp(
        date=now.date().isoformat(),
        full=now.isoformat().replace("+00:00", "Z"),
    )
 
 
def _stamp_from_arg(value: str | None) -> Stamp:
    if value is None:
        return _iso_now()
    date = value.split("T", 1)[0]
    full = value if "T" in value else f"{value}T00:00:00Z"
    return Stamp(date=date, full=full)
 
 
def _run_cmd(
    args: Sequence[str],
    cwd: Path,
    timeout: int = DEFAULT_TIMEOUT_SECONDS,
) -> tuple[int, str, str]:
    """Run a subprocess command with captured output."""
 
    func = "_run_cmd"
    try:
        completed = subprocess.run(
            list(args),
            cwd=str(cwd),
            capture_output=True,
            text=True,
            timeout=timeout,
            check=False,
        )
    except FileNotFoundError as exc:
        _dbg("OC_CHAT.MAIN", func, f"missing command: {args[0]}", 3)
        return 127, "", str(exc)
    except subprocess.TimeoutExpired as exc:
        _dbg("OC_CHAT.MAIN", func, f"timeout: {' '.join(args)}", 3)
        stdout = exc.stdout if isinstance(exc.stdout, str) else ""
        stderr = exc.stderr if isinstance(exc.stderr, str) else ""
        return 124, stdout, stderr or f"timed out after {timeout}s"
    return completed.returncode, completed.stdout, completed.stderr
 
 
def write_branch_inventory(
    repo_root: Path,
    out_dir: Path,
    stamp: Stamp,
    *,
    timeout: int = DEFAULT_TIMEOUT_SECONDS,
) -> Path:
    """Write branch/worktree inventory markdown."""
 
    func = "write_branch_inventory"
    out_dir.mkdir(parents=True, exist_ok=True)
    path = out_dir / f"branch-inventory-{stamp.date}.md"
    branches = _collect_branches(repo_root, timeout)
    worktrees = _collect_worktrees(repo_root, timeout)
    lines = [
        "# M3.1 Branch / Worktree Inventory",
        "",
        f"Generated: {stamp.full}",
        f"Repository: `{repo_root}`",
        "",
        "## Local Branches",
        "",
        "| Branch | Short SHA | Date | Subject | Ahead/Behind vs main | Merged into main? | Disposition |",
        "|---|---|---|---|---:|---|---|",
    ]
    for row in branches:
        lines.append(
            "| "
            + " | ".join(
                [
                    _md(row.name),
                    _md(row.sha),
                    _md(row.date),
                    _md(row.subject),
                    _md(row.ahead_behind),
                    row.merged,
                    "",
                ]
            )
            + " |"
        )
    lines.extend(
        [
            "",
            "## Worktrees",
            "",
            "| Path | Branch | Short SHA | Date | Subject | Ahead/Behind vs main | Merged into main? | Disposition |",
            "|---|---|---|---|---|---:|---|---|",
        ]
    )
    for row in worktrees:
        lines.append(
            "| "
            + " | ".join(
                [
                    _md(row.path),
                    _md(row.branch),
                    _md(row.sha),
                    _md(row.date),
                    _md(row.subject),
                    _md(row.ahead_behind),
                    row.merged,
                    "",
                ]
            )
            + " |"
        )
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    _dbg("OC_CHAT.MAIN", func, f"wrote {path}", 2)
    return path
 
 
def write_health_baseline(
    repo_root: Path,
    out_dir: Path,
    stamp: Stamp,
    *,
    timeout: int = DEFAULT_TIMEOUT_SECONDS,
) -> Path:
    """Write ``oct health --json`` output or a structured SKIPPED stub."""
 
    func = "write_health_baseline"
    out_dir.mkdir(parents=True, exist_ok=True)
    path = out_dir / f"health-baseline-{stamp.date}.json"
    rc, stdout, stderr = _run_cmd(["oct", "health", "--json"], repo_root, timeout)
    if rc == 0 and stdout.strip():
        text = stdout
    else:
        status = "SKIPPED" if rc == 127 else "ERROR"
        text = json.dumps(
            {
                "status": status,
                "generated_at": stamp.full,
                "command": "oct health --json",
                "returncode": rc,
                "reason": (stderr or stdout or "no output").strip(),
            },
            indent=2,
            sort_keys=True,
        )
        text += "\n"
    path.write_text(text, encoding="utf-8")
    _dbg("OC_CHAT.MAIN", func, f"wrote {path}", 2)
    return path
 
 
def write_test_baseline(
    repo_root: Path,
    out_dir: Path,
    stamp: Stamp,
    *,
    timeout: int = DEFAULT_TIMEOUT_SECONDS,
) -> Path:
    """Write pytest collection-count baseline."""
 
    func = "write_test_baseline"
    out_dir.mkdir(parents=True, exist_ok=True)
    path = out_dir / f"test-baseline-{stamp.date}.txt"
    command = [sys.executable, "-m", "pytest", "oc_chat/tests/", "--collect-only", "-q"]
    rc, stdout, stderr = _run_cmd(command, repo_root, timeout)
    combined = "\n".join(part for part in (stdout, stderr) if part)
    count = _parse_test_count(combined)
    lines = [
        "# M3.1 Test Baseline",
        "",
        f"Generated: {stamp.full}",
        f"Command: {' '.join(command)}",
        f"Return code: {rc}",
        f"Collected tests: {count if count is not None else 'UNKNOWN'}",
        "",
        "## Raw Tail",
        "",
        "```text",
        _tail(combined),
        "```",
    ]
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    _dbg("OC_CHAT.MAIN", func, f"wrote {path} count={count}", 2)
    return path
 
 
def write_filter_repo_dryrun(
    repo_root: Path,
    out_dir: Path,
    stamp: Stamp,
    *,
    timeout: int = DEFAULT_TIMEOUT_SECONDS,
) -> Path:
    """Run filter-repo only inside a throwaway clone and write the report."""
 
    func = "write_filter_repo_dryrun"
    out_dir.mkdir(parents=True, exist_ok=True)
    path = out_dir / f"filter-repo-dryrun-{stamp.date}.txt"
    if shutil.which("git-filter-repo") is None:
        _write_filter_report(
            path,
            stamp,
            "SKIPPED",
            "git-filter-repo not available on PATH",
        )
        _dbg("OC_CHAT.MAIN", func, "git-filter-repo unavailable", 2)
        return path
 
    mode = _filter_repo_mode(repo_root, timeout)
    if mode is None:
        _write_filter_report(
            path,
            stamp,
            "SKIPPED",
            "installed git-filter-repo exposes neither --dry-run nor --analyze",
        )
        return path
 
    clone_dir: Path | None = None
    safe_to_remove = False
    try:
        clone_dir = Path(
            tempfile.mkdtemp(
                prefix="oc-chat-filter-repo-",
                dir=tempfile.gettempdir(),
            )
        )
        safe_to_remove = _safe_clone_dir(clone_dir, repo_root)
        if not safe_to_remove:
            _write_filter_report(
                path,
                stamp,
                "ERROR",
                f"refusing unsafe clone dir: {clone_dir}",
            )
            return path
        clone_rc, clone_out, clone_err = _run_cmd(
            ["git", "clone", "--no-hardlinks", str(repo_root), str(clone_dir)],
            Path(tempfile.gettempdir()),
            timeout,
        )
        if clone_rc != 0:
            _write_filter_report(
                path,
                stamp,
                "ERROR",
                "git clone failed",
                command="git clone --no-hardlinks",
                returncode=clone_rc,
                stdout=clone_out,
                stderr=clone_err,
            )
            return path
        command = _filter_repo_command(mode)
        rc, stdout, stderr = _run_cmd(command, clone_dir, timeout)
        _write_filter_report(
            path,
            stamp,
            "OK" if rc == 0 else "ERROR",
            f"git filter-repo completed with rc={rc}",
            command=" ".join(command),
            returncode=rc,
            stdout=stdout,
            stderr=stderr,
        )
        return path
    finally:
        if clone_dir is not None and safe_to_remove:
            shutil.rmtree(clone_dir, ignore_errors=True)
 
 
def run_all(
    repo_root: Path,
    out_dir: Path,
    stamp: Stamp,
    *,
    timeout: int = DEFAULT_TIMEOUT_SECONDS,
) -> list[Path]:
    """Run all pre-flight producers and write the summary."""
 
    func = "run_all"
    artifacts = [
        write_branch_inventory(repo_root, out_dir, stamp, timeout=timeout),
        write_health_baseline(repo_root, out_dir, stamp, timeout=timeout),
        write_test_baseline(repo_root, out_dir, stamp, timeout=timeout),
        write_filter_repo_dryrun(repo_root, out_dir, stamp, timeout=timeout),
    ]
    summary = write_summary(out_dir, stamp, artifacts)
    _dbg("OC_CHAT.MAIN", func, f"wrote {len(artifacts)} artifacts", 2)
    return [*artifacts, summary]
 
 
def write_summary(out_dir: Path, stamp: Stamp, artifacts: Sequence[Path]) -> Path:
    """Write the pre-flight summary markdown."""
 
    out_dir.mkdir(parents=True, exist_ok=True)
    path = out_dir / f"PREFLIGHT-SUMMARY-{stamp.date}.md"
    lines = [
        "# M3.1 Monorepo Migration Pre-flight Summary",
        "",
        f"Generated: {stamp.full}",
        "",
        "## Automated Artifacts",
        "",
    ]
    for artifact in artifacts:
        lines.append(f"- [{artifact.name}]({artifact.name})")
    lines.extend(
        [
            "",
            "## Orchestrator-owned Manual Steps",
            "",
            "- Create tag `v0.7.8` at the approved pre-migration commit.",
            "- Announce the 48-hour branch freeze before M3.2 begins.",
            "- Announce the later M3.8 unfreeze after the migration settles.",
            "",
            "## M3.2 Consumption",
            "",
            "- Use the branch inventory to account for every branch/worktree disposition.",
            "- Compare the health baseline against post-migration `oct health --json`.",
            "- Match the collected test count before accepting the migrated workspace.",
            "- Review the filter-repo dry-run output before the live history rewrite.",
        ]
    )
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    return path
 
 
def _collect_branches(repo_root: Path, timeout: int) -> list[BranchRow]:
    rc, stdout, stderr = _run_cmd(
        [
            "git",
            "for-each-ref",
            "--format=%(refname:short)%09%(objectname:short)%09%(committerdate:short)%09%(contents:subject)",
            "refs/heads",
        ],
        repo_root,
        timeout,
    )
    if rc != 0:
        return [
            BranchRow(
                name="ERROR",
                sha="",
                date="",
                subject=(stderr or stdout or "git for-each-ref failed").strip(),
                ahead_behind="",
                merged="no",
            )
        ]
    merged = _merged_branches(repo_root, timeout)
    rows: list[BranchRow] = []
    for line in stdout.splitlines():
        if not line.strip():
            continue
        name, sha, date, subject = _split_tab4(line)
        ahead_behind = _ahead_behind(repo_root, name, timeout)
        rows.append(
            BranchRow(
                name=name,
                sha=sha,
                date=date,
                subject=subject,
                ahead_behind=ahead_behind,
                merged="yes" if name in merged else "no",
            )
        )
    return rows
 
 
def _collect_worktrees(repo_root: Path, timeout: int) -> list[WorktreeRow]:
    rc, stdout, stderr = _run_cmd(["git", "worktree", "list", "--porcelain"], repo_root, timeout)
    if rc != 0:
        return [
            WorktreeRow(
                path="ERROR",
                branch="",
                sha="",
                date="",
                subject=(stderr or stdout or "git worktree list failed").strip(),
                ahead_behind="",
                merged="no",
            )
        ]
    rows: list[WorktreeRow] = []
    for record in _parse_worktree_records(stdout):
        worktree = record.get("worktree", "")
        branch = record.get("branch", "(detached)").removeprefix("refs/heads/")
        sha, date, subject = _worktree_commit_info(repo_root, worktree, timeout)
        rows.append(
            WorktreeRow(
                path=worktree,
                branch=branch,
                sha=sha or record.get("HEAD", "")[:7],
                date=date,
                subject=subject,
                ahead_behind=_ahead_behind(repo_root, branch, timeout)
                if branch != "(detached)"
                else "",
                merged="yes"
                if branch != "(detached)" and branch in _merged_branches(repo_root, timeout)
                else "no",
            )
        )
    return rows
 
 
def _merged_branches(repo_root: Path, timeout: int) -> set[str]:
    rc, stdout, _stderr = _run_cmd(
        ["git", "branch", "--merged", "main", "--format=%(refname:short)"],
        repo_root,
        timeout,
    )
    if rc != 0:
        return set()
    return {line.strip().lstrip("* ").strip() for line in stdout.splitlines() if line.strip()}
 
 
def _ahead_behind(repo_root: Path, branch: str, timeout: int) -> str:
    rc, stdout, _stderr = _run_cmd(
        ["git", "rev-list", "--left-right", "--count", f"main...{branch}"],
        repo_root,
        timeout,
    )
    if rc != 0:
        return "unknown"
    parts = stdout.strip().split()
    if len(parts) < 2:
        return "unknown"
    behind, ahead = parts[0], parts[1]
    return f"{ahead}/{behind}"
 
 
def _worktree_commit_info(
    repo_root: Path,
    worktree: str,
    timeout: int,
) -> tuple[str, str, str]:
    rc, stdout, stderr = _run_cmd(
        ["git", "-C", worktree, "log", "-1", "--format=%h%x09%cs%x09%s"],
        repo_root,
        timeout,
    )
    if rc != 0:
        return "", "", (stderr or stdout or "git log failed").strip()
    parts = stdout.strip().split("\t", 2)
    while len(parts) < 3:
        parts.append("")
    return parts[0], parts[1], parts[2]
 
 
def _parse_worktree_records(text: str) -> list[dict[str, str]]:
    records: list[dict[str, str]] = []
    current: dict[str, str] = {}
    for line in text.splitlines():
        if not line.strip():
            if current:
                records.append(current)
                current = {}
            continue
        key, _, value = line.partition(" ")
        current[key] = value
    if current:
        records.append(current)
    return records
 
 
def _parse_test_count(text: str) -> int | None:
    matches = COUNT_RE.findall(text)
    if matches:
        return int(matches[-1])
    per_file_counts = [int(value) for value in PER_FILE_COUNT_RE.findall(text)]
    if per_file_counts:
        return sum(per_file_counts)
    return None
 
 
def _filter_repo_mode(repo_root: Path, timeout: int) -> str | None:
    rc, stdout, stderr = _run_cmd(["git", "filter-repo", "--help"], repo_root, timeout)
    if rc not in {0, 1}:
        return None
    help_text = f"{stdout}\n{stderr}"
    if "--dry-run" in help_text:
        return "--dry-run"
    if "--analyze" in help_text:
        return "--analyze"
    return None
 
 
def _filter_repo_command(mode: str) -> list[str]:
    if mode == "--dry-run":
        return ["git", "filter-repo", "--to-subdirectory-filter", "oc_chat/", "--dry-run"]
    return ["git", "filter-repo", "--analyze"]
 
 
def _safe_clone_dir(clone_dir: Path, repo_root: Path) -> bool:
    clone_resolved = clone_dir.resolve()
    repo_resolved = repo_root.resolve()
    temp_resolved = Path(tempfile.gettempdir()).resolve()
    return clone_resolved != repo_resolved and (
        clone_resolved == temp_resolved or temp_resolved in clone_resolved.parents
    )
 
 
def _write_filter_report(
    path: Path,
    stamp: Stamp,
    status: str,
    reason: str,
    *,
    command: str | None = None,
    returncode: int | None = None,
    stdout: str = "",
    stderr: str = "",
) -> None:
    lines = [
        "# M3.1 filter-repo dry-run",
        "",
        f"Generated: {stamp.full}",
        f"Status: {status}",
        f"Reason: {reason}",
    ]
    if command is not None:
        lines.extend(["", f"Command: {command}", f"Return code: {returncode}"])
    if stdout:
        lines.extend(["", "## stdout", "", "```text", stdout.rstrip(), "```"])
    if stderr:
        lines.extend(["", "## stderr", "", "```text", stderr.rstrip(), "```"])
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
 
 
def _split_tab4(line: str) -> tuple[str, str, str, str]:
    parts = line.split("\t", 3)
    while len(parts) < 4:
        parts.append("")
    return parts[0], parts[1], parts[2], parts[3]
 
 
def _tail(text: str, lines: int = 25) -> str:
    return "\n".join(text.splitlines()[-lines:])
 
 
def _md(value: str) -> str:
    return str(value).replace("|", "\\|").replace("\n", " ").strip()
 
 
def _resolve_repo_root(raw: str | None) -> Path:
    return Path(raw or ".").resolve()
 
 
def _resolve_out_dir(repo_root: Path, raw: str | None) -> Path:
    candidate = Path(raw or DEFAULT_OUT_DIR)
    return candidate.resolve() if candidate.is_absolute() else (repo_root / candidate).resolve()
 
 
def _add_common(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--repo-root", help="Repository root. Defaults to cwd.")
    parser.add_argument("--out-dir", default=DEFAULT_OUT_DIR, help="Artifact output directory.")
    parser.add_argument("--stamp", help="Override artifact ISO date/full timestamp for tests.")
    parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT_SECONDS)
 
 
def _dispatch(args: argparse.Namespace) -> list[Path]:
    repo_root = _resolve_repo_root(args.repo_root)
    out_dir = _resolve_out_dir(repo_root, args.out_dir)
    stamp = _stamp_from_arg(args.stamp)
    if args.command == "branch-inventory":
        return [write_branch_inventory(repo_root, out_dir, stamp, timeout=args.timeout)]
    if args.command == "health-baseline":
        return [write_health_baseline(repo_root, out_dir, stamp, timeout=args.timeout)]
    if args.command == "test-baseline":
        return [write_test_baseline(repo_root, out_dir, stamp, timeout=args.timeout)]
    if args.command == "filter-repo-dryrun":
        return [write_filter_repo_dryrun(repo_root, out_dir, stamp, timeout=args.timeout)]
    if args.command == "run":
        return run_all(repo_root, out_dir, stamp, timeout=args.timeout)
    raise ValueError(f"unknown command: {args.command}")
 
 
def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="python -m oc_chat.scripts.migration_preflight",
        description="Capture M3.1 monorepo migration pre-flight artifacts.",
    )
    sub = parser.add_subparsers(dest="command", required=True)
    for name, help_text in (
        ("run", "Run every pre-flight producer and write a summary."),
        ("branch-inventory", "Write branch/worktree inventory markdown."),
        ("health-baseline", "Capture oct health --json."),
        ("test-baseline", "Record pytest collection count."),
        ("filter-repo-dryrun", "Run filter-repo in a temp clone or record SKIPPED."),
    ):
        child = sub.add_parser(name, help=help_text)
        _add_common(child)
    return parser
 
 
def main(argv: list[str] | None = None) -> int:
    func = "main"
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        paths = _dispatch(args)
    except Exception as exc:  # noqa: BLE001 - CLI boundary
        print(f"migration preflight failed: {type(exc).__name__}: {exc}", file=sys.stderr)
        return 1
    for path in paths:
        print(path)
    _dbg("OC_CHAT.MAIN", func, f"completed {args.command}", 2)
    return 0
 
 
if __name__ == "__main__":
    sys.exit(main())