migrate.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/cli/migrate.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_migration.py"
"""
Migration operator commands for the orchestration registry.
 
Purpose
-------
Expose the legacy Markdown-state migration lifecycle beneath
``python -m oc_chat.registry migrate``.
 
Responsibilities
----------------
- Wire import, verify, rollback, and cutover subcommands.
- Resolve registry/workspace paths through the standard discovery logic.
- Format migration results as human text or JSON.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- ``cutover`` is a v0.2.x stub and exits 2.
- ``verify`` exits 1 when the registry differs from the legacy projection.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.migrate (migration command implementations)
- oc_chat.registry.discovery (registry path discovery)
- oc_chat.registry (Registry facade passed to the migration commands)
- oc_chat.registry.backends.sqlite (SQLiteBackend opened at the discovered registry path)
"""
from __future__ import annotations
 
import argparse
import json
import sys
from pathlib import Path
from typing import Mapping
 
from oc_diagnostics import _dbg
 
from oc_chat.registry import Registry
from oc_chat.registry.backends.sqlite import SQLiteBackend
from oc_chat.registry.discovery import discover_registry_location
from oc_chat.registry.migrate import (
    MigrationOptions,
    run_cutover_stub,
    run_import,
    run_rollback,
    run_verify,
)
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
def add_migrate_subcommands(migrate_parser: argparse.ArgumentParser) -> None:
    """Attach the M11.2 migration subcommands."""
 
    migrate_parser.set_defaults(handler=_cmd_migrate_group, migrate_parser=migrate_parser)
    subcommands = migrate_parser.add_subparsers(dest="migrate_command")
 
    import_parser = subcommands.add_parser(
        "import",
        help="Import legacy queue/status files into the registry.",
    )
    import_parser.add_argument(
        "registry",
        nargs="?",
        help="Registry SQLite path. Defaults to the discovery algorithm.",
    )
    import_parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Build and report the projection without writing.",
    )
    import_parser.add_argument(
        "--json",
        action="store_true",
        help="Emit the structured import report as JSON.",
    )
    import_parser.set_defaults(handler=_cmd_import)
 
    verify_parser = subcommands.add_parser(
        "verify",
        help="Compare the registry with the current legacy projection.",
    )
    verify_parser.add_argument(
        "registry",
        nargs="?",
        help="Registry SQLite path. Defaults to the discovery algorithm.",
    )
    verify_parser.add_argument(
        "--json",
        action="store_true",
        help="Emit the structured verify report as JSON.",
    )
    verify_parser.set_defaults(handler=_cmd_verify)
 
    rollback_parser = subcommands.add_parser(
        "rollback",
        help="Restore the most recent pre-migration registry backup.",
    )
    rollback_parser.add_argument(
        "registry",
        nargs="?",
        help="Registry SQLite path. Defaults to the discovery algorithm.",
    )
    rollback_parser.add_argument(
        "--json",
        action="store_true",
        help="Emit the structured rollback report as JSON.",
    )
    rollback_parser.set_defaults(handler=_cmd_rollback)
 
    cutover_parser = subcommands.add_parser(
        "cutover",
        help="Reserved for the M11.8 cutover phase.",
    )
    cutover_parser.add_argument(
        "--json",
        action="store_true",
        help="Emit the structured cutover report as JSON.",
    )
    cutover_parser.set_defaults(handler=_cmd_cutover)
 
 
def _cmd_migrate_group(args: argparse.Namespace) -> int:
    parser = args.migrate_parser
    parser.print_help()
    return 1
 
 
def _cmd_import(args: argparse.Namespace) -> int:
    registry, options = _registry_and_options(args.registry)
    result = run_import(registry, options, dry_run=args.dry_run)
    _print_result(result.to_dict(), args.json, "migration import")
    return 0 if result.ok else 1
 
 
def _cmd_verify(args: argparse.Namespace) -> int:
    registry, options = _registry_and_options(args.registry)
    result = run_verify(registry, options)
    _print_result(result.to_dict(), args.json, "migration verify")
    return 0 if result.ok else 1
 
 
def _cmd_rollback(args: argparse.Namespace) -> int:
    registry, options = _registry_and_options(args.registry)
    result = run_rollback(registry, options)
    _print_result(result.to_dict(), args.json, "migration rollback")
    return 0 if result.ok else 1
 
 
def _cmd_cutover(args: argparse.Namespace) -> int:
    result = run_cutover_stub()
    if args.json:
        print(json.dumps(result.to_dict(), indent=2, sort_keys=True))
    else:
        print(result.message, file=sys.stderr)
    return result.exit_code
 
 
def _registry_and_options(
    registry_path: str | None,
) -> tuple[Registry, MigrationOptions]:
    location = discover_registry_location(registry_path=registry_path)
    backend = SQLiteBackend(location.registry_path)
    return (
        Registry(backend),
        MigrationOptions(workspace_root=Path(location.workspace_root)),
    )
 
 
def _print_result(
    result: Mapping[str, object],
    as_json: bool,
    label: str,
) -> None:
    if as_json:
        print(json.dumps(result, indent=2, sort_keys=True))
        return
    print(format_migrate_report(result, label))
 
 
def format_migrate_report(result: Mapping[str, object], label: str) -> str:
    """Return a compact human-readable migration report."""
 
    ok = result.get("ok") is True
    summary = result.get("summary", {})
    lines = [f"{label} {'ok' if ok else 'failed'}"]
    if isinstance(summary, Mapping):
        counts = ", ".join(
            f"{key}={value}"
            for key, value in summary.items()
            if key != "status_counts"
        )
        if counts:
            lines.append(f"  {counts}")
        status_counts = summary.get("status_counts")
        if isinstance(status_counts, Mapping):
            rendered = ", ".join(f"{key}={value}" for key, value in status_counts.items())
            lines.append(f"  statuses: {rendered}")
    backup_path = result.get("backup_path") or result.get("restored_from")
    if backup_path:
        lines.append(f"  backup: {backup_path}")
    for field in ("warnings", "errors"):
        values = result.get(field, [])
        if isinstance(values, list) and values:
            lines.append(f"  {field}:")
            lines.extend(f"    - {value}" for value in values)
    if result.get("diff"):
        lines.append("  diff: present")
    return "\n".join(lines)