#!/usr/bin/env python3 """ Migrate User Data - Separate user data from framework (ADR-114)

Migrates session-logs, context-storage, and machine-id.json from the core/ directory to a sibling data/ directory.

Usage: python3 scripts/migrate-user-data.py # Run migration python3 scripts/migrate-user-data.py --dry-run # Preview only python3 scripts/migrate-user-data.py --status # Check current state python3 scripts/migrate-user-data.py --rollback # Revert migration

Part of ADR-114: User Data Separation from Framework

Created: 2026-01-25 """

import argparse import json import os import shutil import sys from datetime import datetime from pathlib import Path from typing import Optional, Tuple, List

ANSI colors

Shared Colors module (consolidates 36 duplicate definitions)

_script_dir = Path(file).parent sys.path.insert(0, str(_script_dir / "core")) from colors import Colors

def colorize(text: str, color: str) -> str: return f"{color}{text}{Colors.END}"

Import from shared paths module (ADR-114)

This handles configurable PROJECTS location for customer flexibility

try: from scripts.core.paths import ( discover_projects_dir, get_framework_dir, get_user_data_dir, HOME, ) PROJECTS_DIR = discover_projects_dir() FRAMEWORK_LOC = get_framework_dir() USER_DATA_LOC = get_user_data_dir() except ImportError: # Fallback if paths module not available (standalone execution) HOME = Path.home()

# Discover PROJECTS directory
def _discover_projects_dir() -> Path:
    """Discover PROJECTS dir - see scripts/core/paths.py for full implementation."""
    # 1. Environment variable
    if env_projects := os.environ.get("CODITECT_PROJECTS"):
        return Path(env_projects).expanduser()

    # 2. Config file
    config_path = HOME / ".coditect" / "config" / "config.json"
    if config_path.exists():
        try:
            with open(config_path) as f:
                config = json.load(f)
                if projects_dir := config.get("projects_dir"):
                    return Path(projects_dir).expanduser()
        except (json.JSONDecodeError, IOError):
            pass

    # 3. Symlink discovery
    for candidate in [HOME / "PROJECTS", HOME / "Projects", HOME / "Dev",
                      HOME / "Development", HOME / "Code"]:
        if candidate.exists() and (candidate / ".coditect").is_symlink():
            return candidate

    # 4. Default
    return HOME / "PROJECTS"

PROJECTS_DIR = _discover_projects_dir()

# Framework location (protected, synced from GitHub)
if sys.platform == "darwin":
    CODITECT_BASE = HOME / "Library" / "Application Support" / "CODITECT"
elif sys.platform == "win32":
    CODITECT_BASE = Path(os.environ.get("LOCALAPPDATA", HOME / "AppData" / "Local")) / "CODITECT"
else:
    CODITECT_BASE = Path(os.environ.get("XDG_DATA_HOME", HOME / ".local" / "share")) / "coditect"

FRAMEWORK_LOC = CODITECT_BASE / "core"
USER_DATA_LOC = PROJECTS_DIR / ".coditect-data"

Old location for migration detection (legacy)

CODITECT_BASE = FRAMEWORK_LOC.parent # ~/Library/.../CODITECT/ OLD_USER_DATA_LOC = CODITECT_BASE / "data"

Files/directories to migrate

USER_DATA_ITEMS = [ "machine-id.json", "session-logs", "context-storage", ]

Symlinks to create/update

SYMLINKS = { HOME / ".coditect": FRAMEWORK_LOC, HOME / ".coditect-data": USER_DATA_LOC, }

def get_migration_status() -> dict: """Check current migration status.""" status = { "framework_exists": FRAMEWORK_LOC.exists(), "user_data_exists": USER_DATA_LOC.exists(), "old_data_exists": OLD_USER_DATA_LOC.exists(), "items_in_core": [], "items_in_old_data": [], # Items in ~/Library/.../CODITECT/data/ "items_in_new_data": [], # Items in ~/PROJECTS/.coditect-data/ "symlinks_to_old": [], # Symlinks pointing to old location "symlinks_to_new": [], # Symlinks pointing to new location "symlinks_external": [], # Symlinks pointing outside (e.g., git repo) "migration_complete": False, }

for item in USER_DATA_ITEMS:
    core_path = FRAMEWORK_LOC / item
    old_data_path = OLD_USER_DATA_LOC / item
    new_data_path = USER_DATA_LOC / item

    # Check what's in core/
    if core_path.exists() and not core_path.is_symlink():
        status["items_in_core"].append(item)
    elif core_path.is_symlink():
        try:
            target = core_path.resolve()
            if str(target).startswith(str(USER_DATA_LOC)):
                status["symlinks_to_new"].append(item)
            elif str(target).startswith(str(OLD_USER_DATA_LOC)):
                status["symlinks_to_old"].append((item, str(target)))
            else:
                # External symlink (e.g., session-logs → git repo)
                status["symlinks_external"].append((item, str(target)))
        except OSError:
            pass

    # Check old data location
    if old_data_path.exists() and not old_data_path.is_symlink():
        status["items_in_old_data"].append(item)

    # Check new data location
    if new_data_path.exists() and not new_data_path.is_symlink():
        status["items_in_new_data"].append(item)
    elif new_data_path.is_symlink():
        # Symlink in new location (e.g., session-logs → git repo)
        try:
            target = new_data_path.resolve()
            if (item, str(target)) not in status["symlinks_external"]:
                status["symlinks_external"].append((item, str(target)))
        except OSError:
            pass

# Migration is complete if:
# - All items are in new location (or symlinked there)
# - No items remain in old location or core/
items_in_new = len(status["items_in_new_data"]) + len(status["symlinks_to_new"])
# Count external symlinks that are also in new location
external_in_new = sum(1 for item, _ in status["symlinks_external"]
                      if (USER_DATA_LOC / item).exists())
items_handled = items_in_new + external_in_new

status["migration_complete"] = (
    items_handled >= len(USER_DATA_ITEMS) and
    len(status["items_in_core"]) == 0 and
    len(status["items_in_old_data"]) == 0 and
    len(status["symlinks_to_old"]) == 0
)

return status

def print_status(): """Print current migration status.""" status = get_migration_status()

print(colorize("\n╔══════════════════════════════════════════════════════════════╗", Colors.CYAN))
print(colorize("║           CODITECT User Data Migration Status                 ║", Colors.CYAN))
print(colorize("╚══════════════════════════════════════════════════════════════╝", Colors.CYAN))

print(f"\nFramework location: {FRAMEWORK_LOC}")
print(f"  Exists: {'✓' if status['framework_exists'] else '✗'}")

print(f"\nOld data location: {OLD_USER_DATA_LOC}")
print(f"  Exists: {'✓' if status['old_data_exists'] else '✗'}")

print(f"\nNew data location: {USER_DATA_LOC}")
print(f"  Exists: {'✓' if status['user_data_exists'] else '✗'}")

print("\nUser data items:")
external_symlinks = {item: target for item, target in status.get("symlinks_external", [])}
symlinks_to_old = {item: target for item, target in status.get("symlinks_to_old", [])}

for item in USER_DATA_ITEMS:
    in_core = item in status["items_in_core"]
    in_old_data = item in status["items_in_old_data"]
    in_new_data = item in status["items_in_new_data"]
    symlink_to_new = item in status["symlinks_to_new"]
    symlink_to_old = item in symlinks_to_old
    is_external = item in external_symlinks

    new_data_path = USER_DATA_LOC / item
    has_symlink_in_new = new_data_path.is_symlink()

    if in_new_data:
        icon = colorize("✓", Colors.GREEN)
        loc = f".coditect-data/ (migrated)"
    elif has_symlink_in_new and is_external:
        target = external_symlinks[item]
        short_target = target.replace(str(HOME), "~")
        icon = colorize("✓", Colors.GREEN)
        loc = f".coditect-data/ → {short_target}"
    elif symlink_to_new:
        icon = colorize("✓", Colors.GREEN)
        loc = "core/ → .coditect-data/"
    elif in_old_data or symlink_to_old:
        icon = colorize("!", Colors.YELLOW)
        loc = "CODITECT/data/ (needs migration)"
    elif in_core:
        icon = colorize("!", Colors.YELLOW)
        loc = "core/ (needs migration)"
    elif is_external and not has_symlink_in_new:
        target = external_symlinks[item]
        short_target = target.replace(str(HOME), "~")
        icon = colorize("↗", Colors.YELLOW)
        loc = f"external only → {short_target} (needs symlink in new location)"
    else:
        icon = colorize("-", Colors.YELLOW)
        loc = "missing"

    print(f"  {icon} {item}: {loc}")

print()
if status["migration_complete"]:
    print(colorize("Migration Status: COMPLETE ✓", Colors.GREEN))
else:
    print(colorize("Migration Status: PENDING", Colors.YELLOW))
    if status["items_in_old_data"] or status["symlinks_to_old"]:
        print(f"  Data in old location needs migration to: {USER_DATA_LOC}")
    print("  Run: python3 scripts/migrate-user-data.py")

def backup_item(path: Path, backup_dir: Path) -> Optional[Path]: """Create a backup of an item.""" if not path.exists(): return None

backup_path = backup_dir / f"{path.name}.backup.{datetime.now().strftime('%Y%m%d_%H%M%S')}"

if path.is_dir():
    shutil.copytree(path, backup_path)
else:
    shutil.copy2(path, backup_path)

return backup_path

def migrate_item(item: str, dry_run: bool = False) -> Tuple[bool, str]: """Migrate a single item to PROJECTS/.coditect-data/.

Checks multiple source locations:
1. core/ (framework directory) - original location
2. CODITECT/data/ (old ADR-114 location) - intermediate location
3. External symlinks (e.g., session-logs → git repo)
"""
core_path = FRAMEWORK_LOC / item
old_data_path = OLD_USER_DATA_LOC / item  # ~/Library/.../CODITECT/data/
new_data_path = USER_DATA_LOC / item      # ~/PROJECTS/.coditect-data/

# Skip if already in new location
if new_data_path.exists() and not new_data_path.is_symlink():
    return True, f"Already in new location: {item}"

# Check if core_path is a symlink to external location (e.g., session-logs → git repo)
if core_path.is_symlink():
    target = core_path.resolve()
    # If symlink points to old_data_path, we need to migrate
    if str(target).startswith(str(OLD_USER_DATA_LOC)):
        # Symlink to old location - need to move the actual data
        source_path = target
        if not source_path.exists():
            return True, f"Symlink target missing: {item}"
    elif str(target).startswith(str(USER_DATA_LOC)):
        # Already points to new location
        return True, f"Already symlinked to new location: {item}"
    else:
        # External symlink (e.g., session-logs → git repo)
        # Create symlink in new location pointing to same target
        if not new_data_path.exists():
            if dry_run:
                return True, f"Would create symlink in new location: {item} → {target}"
            try:
                new_data_path.symlink_to(target)
                return True, f"Created symlink in new location: {item}"
            except Exception as e:
                return False, f"Failed to create symlink: {e}"
        return True, f"External symlink preserved: {item}"
else:
    source_path = None

# Find the actual data source
if source_path is None:
    if old_data_path.exists() and not old_data_path.is_symlink():
        source_path = old_data_path
    elif core_path.exists() and not core_path.is_symlink():
        source_path = core_path
    else:
        return True, f"No data found to migrate: {item}"

# Handle conflict
if new_data_path.exists():
    return False, f"Conflict: {item} exists in both old and new locations"

if dry_run:
    return True, f"Would migrate: {item} from {source_path.parent.name}/"

try:
    # Move data to new location
    if source_path.is_dir():
        shutil.copytree(str(source_path), str(new_data_path))
        shutil.rmtree(str(source_path))
    else:
        shutil.copy2(str(source_path), str(new_data_path))
        source_path.unlink()

    # Update symlink in core/ to point to new location
    if core_path.is_symlink():
        core_path.unlink()
    core_path.symlink_to(new_data_path)

    # Remove old data location entry if it was a symlink
    if old_data_path.is_symlink():
        old_data_path.unlink()

    return True, f"Migrated: {item} → {USER_DATA_LOC.name}/"
except Exception as e:
    return False, f"Failed to migrate {item}: {e}"

def create_symlinks(dry_run: bool = False) -> List[Tuple[bool, str]]: """Create/update global symlinks.""" results = []

for symlink_path, target_path in SYMLINKS.items():
    if symlink_path.exists() or symlink_path.is_symlink():
        # Check if it already points to the right place
        if symlink_path.is_symlink():
            current_target = symlink_path.resolve()
            if current_target == target_path.resolve():
                results.append((True, f"Symlink OK: {symlink_path.name}"))
                continue

        if dry_run:
            results.append((True, f"Would update symlink: {symlink_path.name}"))
            continue

        # Remove existing
        if symlink_path.is_symlink():
            symlink_path.unlink()
        elif symlink_path.is_dir():
            # Don't delete directories automatically
            results.append((False, f"Cannot replace directory: {symlink_path}"))
            continue

    if dry_run:
        results.append((True, f"Would create symlink: {symlink_path.name} → {target_path}"))
        continue

    try:
        symlink_path.symlink_to(target_path)
        results.append((True, f"Created symlink: {symlink_path.name}"))
    except Exception as e:
        results.append((False, f"Failed to create symlink {symlink_path.name}: {e}"))

return results

def run_migration(dry_run: bool = False): """Run the full migration.""" print(colorize("\n╔══════════════════════════════════════════════════════════════╗", Colors.CYAN)) print(colorize("║ CODITECT User Data Migration (ADR-114) ║", Colors.CYAN)) print(colorize("╚══════════════════════════════════════════════════════════════╝", Colors.CYAN))

if dry_run:
    print(colorize("\n[DRY RUN MODE - No changes will be made]\n", Colors.YELLOW))

# Step 1: Check prerequisites
print(colorize("\n=== Step 1: Check Prerequisites ===", Colors.BOLD))

if not FRAMEWORK_LOC.exists():
    print(colorize(f"✗ Framework not found: {FRAMEWORK_LOC}", Colors.RED))
    print("  Run: python3 scripts/CODITECT-CORE-INITIAL-SETUP.py")
    sys.exit(1)

print(colorize(f"✓ Framework found: {FRAMEWORK_LOC}", Colors.GREEN))

# Step 2: Create data/ directory
print(colorize("\n=== Step 2: Create User Data Directory ===", Colors.BOLD))

if USER_DATA_LOC.exists():
    print(colorize(f"✓ User data directory exists: {USER_DATA_LOC}", Colors.GREEN))
else:
    if dry_run:
        print(colorize(f"  Would create: {USER_DATA_LOC}", Colors.YELLOW))
    else:
        USER_DATA_LOC.mkdir(parents=True, exist_ok=True)
        print(colorize(f"✓ Created: {USER_DATA_LOC}", Colors.GREEN))

# Step 3: Create backup
print(colorize("\n=== Step 3: Backup Existing Data ===", Colors.BOLD))

backup_dir = CODITECT_BASE / "backups" / f"migration-{datetime.now().strftime('%Y%m%d_%H%M%S')}"
if not dry_run:
    backup_dir.mkdir(parents=True, exist_ok=True)
    print(colorize(f"✓ Backup directory: {backup_dir}", Colors.GREEN))

for item in USER_DATA_ITEMS:
    core_path = FRAMEWORK_LOC / item
    if core_path.exists() and not core_path.is_symlink():
        if dry_run:
            print(f"  Would backup: {item}")
        else:
            backup_path = backup_item(core_path, backup_dir)
            if backup_path:
                print(f"  ✓ Backed up: {item}")

# Step 4: Migrate items
print(colorize("\n=== Step 4: Migrate User Data ===", Colors.BOLD))

all_success = True
for item in USER_DATA_ITEMS:
    success, message = migrate_item(item, dry_run)
    icon = colorize("✓", Colors.GREEN) if success else colorize("✗", Colors.RED)
    print(f"  {icon} {message}")
    if not success:
        all_success = False

# Step 5: Update symlinks
print(colorize("\n=== Step 5: Update Global Symlinks ===", Colors.BOLD))

symlink_results = create_symlinks(dry_run)
for success, message in symlink_results:
    icon = colorize("✓", Colors.GREEN) if success else colorize("✗", Colors.RED)
    print(f"  {icon} {message}")
    if not success:
        all_success = False

# Summary
print()
if all_success:
    if dry_run:
        print(colorize("Dry run complete. Run without --dry-run to apply changes.", Colors.YELLOW))
    else:
        print(colorize("╔══════════════════════════════════════════════════════════════╗", Colors.GREEN))
        print(colorize("║                 Migration Complete! ✓                         ║", Colors.GREEN))
        print(colorize("╚══════════════════════════════════════════════════════════════╝", Colors.GREEN))
        print(f"\n  Framework: {FRAMEWORK_LOC}")
        print(f"  User Data: {USER_DATA_LOC}")
        print(f"  Backup:    {backup_dir}")
else:
    print(colorize("Migration completed with errors. Check messages above.", Colors.YELLOW))
    sys.exit(1)

def rollback_migration(dry_run: bool = False): """Rollback migration by moving data back to core/.""" print(colorize("\n╔══════════════════════════════════════════════════════════════╗", Colors.CYAN)) print(colorize("║ CODITECT Migration Rollback ║", Colors.CYAN)) print(colorize("╚══════════════════════════════════════════════════════════════╝", Colors.CYAN))

if dry_run:
    print(colorize("\n[DRY RUN MODE - No changes will be made]\n", Colors.YELLOW))

for item in USER_DATA_ITEMS:
    core_path = FRAMEWORK_LOC / item
    data_path = USER_DATA_LOC / item

    # Remove symlink in core/ if exists
    if core_path.is_symlink():
        if dry_run:
            print(f"  Would remove symlink: {item}")
        else:
            core_path.unlink()
            print(f"  ✓ Removed symlink: {item}")

    # Move from data/ back to core/
    if data_path.exists() and not core_path.exists():
        if dry_run:
            print(f"  Would move back: {item}")
        else:
            shutil.move(str(data_path), str(core_path))
            print(f"  ✓ Moved back: {item}")

print(colorize("\nRollback complete.", Colors.GREEN))

def main(): parser = argparse.ArgumentParser( description="Migrate user data from core/ to data/ (ADR-114)", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: python3 scripts/migrate-user-data.py # Run migration python3 scripts/migrate-user-data.py --dry-run # Preview only python3 scripts/migrate-user-data.py --status # Check status python3 scripts/migrate-user-data.py --rollback # Revert migration """ ) parser.add_argument("--dry-run", action="store_true", help="Preview without making changes") parser.add_argument("--status", action="store_true", help="Show current migration status") parser.add_argument("--rollback", action="store_true", help="Rollback migration")

args = parser.parse_args()

if args.status:
    print_status()
elif args.rollback:
    rollback_migration(args.dry_run)
else:
    run_migration(args.dry_run)

if name == "main": main()