""" Migration Engine for Loki Mode. Core backend for the `loki migrate` enterprise code transformation feature. Implements data models, MigrationPipeline, and phase gates for safe, incremental codebase migrations with checkpoint/rollback support. """ from __future__ import annotations import contextlib import dataclasses import json import logging import os import re import secrets import subprocess import tempfile import threading from dataclasses import asdict, dataclass, field, fields from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional try: import fcntl # POSIX only (macOS + Linux). Absent on Windows. _HAS_FCNTL = True except ImportError: # pragma: no cover - non-POSIX fallback fcntl = None # type: ignore[assignment] _HAS_FCNTL = False logger = logging.getLogger("loki-migration") LOKI_DATA_DIR = os.environ.get("LOKI_DATA_DIR", os.path.expanduser("~/.loki")) MIGRATIONS_DIR = os.path.join(LOKI_DATA_DIR, "migrations") # Phase ordering for gate validation PHASE_ORDER = ["understand", "guardrail", "migrate", "verify"] # --------------------------------------------------------------------------- # Data Models # --------------------------------------------------------------------------- @dataclass class Feature: """Individual feature tracked during migration.""" id: str category: str = "" description: str = "" verification_steps: list[str] = field(default_factory=list) passes: bool = False characterization_test: str = "" risk: str = "low" notes: str = "" @dataclass class MigrationStep: """Single step in a migration plan.""" id: str description: str = "" type: str = "" # e.g. "refactor", "rewrite", "config", "test" files: list[str] = field(default_factory=list) tests_required: list[str] = field(default_factory=list) estimated_tokens: int = 0 risk: str = "low" rollback_point: bool = False depends_on: list[str] = field(default_factory=list) assigned_agent: str = "" status: str = "pending" # pending | in_progress | completed | failed @dataclass class MigrationPlan: """Full migration plan with strategy and steps.""" version: int = 1 strategy: str = "incremental" constraints: list[str] = field(default_factory=list) steps: list[MigrationStep] = field(default_factory=list) rollback_strategy: str = "checkpoint" exit_criteria: dict[str, Any] = field(default_factory=dict) @dataclass class SeamInfo: """Detected seam (boundary/interface) in the codebase.""" id: str description: str = "" type: str = "" # e.g. "api", "module", "database", "config" location: str = "" name: str = "" priority: str = "medium" files: list[str] = field(default_factory=list) dependencies: list[str] = field(default_factory=list) complexity: str = "" confidence: float = 0.0 suggested_interface: str = "" @dataclass class PhaseResult: """Result of executing a migration phase.""" phase: str status: str # pending | in_progress | completed | failed artifacts: list[str] = field(default_factory=list) started_at: str = "" completed_at: str = "" error: str = "" @dataclass class CostEstimate: """Token cost estimation for migration.""" total_tokens: int = 0 estimated_cost_usd: float = 0.0 by_phase: dict[str, int] = field(default_factory=dict) @dataclass class MigrationManifest: """Tracks overall migration state.""" id: str = "" created_at: str = "" source_info: dict[str, Any] = field(default_factory=dict) target_info: dict[str, Any] = field(default_factory=dict) phases: dict[str, dict[str, Any]] = field(default_factory=dict) feature_list_path: str = "" migration_plan_path: str = "" checkpoints: list[str] = field(default_factory=list) status: str = "pending" progress_pct: int = 0 updated_at: str = "" source_path: str = "" # --------------------------------------------------------------------------- # Atomic file write helper # --------------------------------------------------------------------------- def _atomic_write(path: Path, content: str) -> None: """Write content to file atomically using temp-file-then-rename. Matches the pattern in prompt_optimizer.py for POSIX safety. """ path.parent.mkdir(parents=True, exist_ok=True) tmp_path = None try: fd, tmp_path = tempfile.mkstemp( dir=str(path.parent), suffix=".tmp" ) try: os.write(fd, content.encode("utf-8")) os.fsync(fd) finally: os.close(fd) os.replace(tmp_path, str(path)) except OSError as exc: logger.error("Failed to write %s: %s", path, exc) # Clean up temp file on failure if tmp_path is not None: try: os.unlink(tmp_path) except OSError: pass raise def _timestamp_iso() -> str: """Return current UTC timestamp in ISO format.""" return datetime.now(timezone.utc).isoformat() @contextlib.contextmanager def _manifest_file_lock(migration_dir: Path): """Cross-process advisory lock around a migration's manifest read-modify-write. Uses an OS file lock (fcntl.flock LOCK_EX) on a dedicated lockfile inside the migration directory. A fresh file descriptor is opened on every acquisition: flock keys on the open file description, so distinct fds let the kernel serialize even threads in the same process (the FastAPI sync threadpool deployment) AND separate processes (the server vs. the `loki migrate` CLI), which a per-instance threading.Lock cannot do. Only the OUTERMOST read-modify-write entry point should take this lock. Nesting two flock-wrapped calls in the same thread would self-deadlock (two fds, the second LOCK_EX blocks on the first), so locked writers must call the _unlocked internals and never re-enter a flock-wrapped method. On non-POSIX platforms (no fcntl) this degrades to a no-op: the caller's in-process threading.Lock still serializes threads in the same process, but cross-process exclusion is NOT available. This is an accepted residual on Windows only. """ migration_dir.mkdir(parents=True, exist_ok=True) lock_path = migration_dir / "manifest.json.lock" if not _HAS_FCNTL: # Graceful degrade: no OS lock available. In-process callers still rely # on their threading.Lock; cross-process safety is unavailable here. yield return fd = os.open(str(lock_path), os.O_RDWR | os.O_CREAT, 0o644) try: fcntl.flock(fd, fcntl.LOCK_EX) try: yield finally: fcntl.flock(fd, fcntl.LOCK_UN) finally: os.close(fd) # --------------------------------------------------------------------------- # MigrationPipeline # --------------------------------------------------------------------------- class MigrationPipeline: """Manages the lifecycle of a codebase migration. All state is persisted under ~/.loki/migrations//. Concurrency: manifest read-modify-write operations (start_phase, advance_phase, save_manifest, update_progress, create_checkpoint) are serialized across BOTH threads and processes by an OS file lock (see _manifest_file_lock) keyed on the migration directory. This holds for the FastAPI sync-endpoint threadpool (each request builds a fresh pipeline via load(), so per-instance threading.Lock alone would not serialize them) and for the separate `loki migrate` CLI process running concurrently with the server. Pure reads use the per-instance lock only. On non-POSIX platforms (no fcntl) the file lock degrades to a no-op and only in-process serialization remains (see _manifest_file_lock). """ def __init__( self, codebase_path: str, target: str, options: Optional[dict[str, Any]] = None, ) -> None: self.codebase_path = os.path.abspath(codebase_path.rstrip(os.sep)) basename = os.path.basename(self.codebase_path) if not basename: raise ValueError(f"Cannot derive project name from codebase path: {codebase_path}") self.target = target self.options = options or {} self.migration_id = self._generate_migration_id() self.migration_dir = Path(MIGRATIONS_DIR) / self.migration_id self._lock = threading.Lock() self._logger = logging.getLogger("loki-migration") # Ensure directory structure exists self.migration_dir.mkdir(parents=True, exist_ok=True) (self.migration_dir / "docs").mkdir(exist_ok=True) (self.migration_dir / "checkpoints").mkdir(exist_ok=True) def _generate_migration_id(self) -> str: """Generate a unique migration ID like mig_20260223_143052_-a1b2c3. The trailing 6-hex-char random suffix prevents collisions when two migrations of the same path-basename start in the same second (the date_str/time_str are second-resolution). Without it, the two would derive the same id and the second create_manifest would overwrite the first (the server rate-limiter throttles same-second server starts, but the CLI bypasses it). The suffix is appended WITHIN the trailing name segment (hyphen-joined), so the load() validation regex ^mig_\\d{8}_\\d{6}_[a-zA-Z0-9_-]+$ still matches without modification (its trailing group already permits letters, digits, hyphens). """ dirname = os.path.basename(self.codebase_path) # Sanitize dirname to match validation regex safe_dirname = re.sub(r'[^a-zA-Z0-9_-]', '_', dirname) if not safe_dirname: safe_dirname = 'unnamed' now = datetime.now(timezone.utc) date_str = now.strftime("%Y%m%d") time_str = now.strftime("%H%M%S") suffix = secrets.token_hex(3) # 6 hex chars, within [a-zA-Z0-9_-]+ return f"mig_{date_str}_{time_str}_{safe_dirname}-{suffix}" @classmethod def load(cls, migration_id: str) -> 'MigrationPipeline': """Load an existing migration by ID.""" if not re.match(r'^mig_\d{8}_\d{6}_[a-zA-Z0-9_-]+$', migration_id): raise ValueError( f"Invalid migration_id '{migration_id}': must match ^mig_YYYYMMDD_HHMMSS_$" ) migrations_dir = MIGRATIONS_DIR migration_dir = os.path.join(migrations_dir, migration_id) if not os.path.isdir(migration_dir): raise FileNotFoundError(f"Migration not found: {migration_id}") manifest_path = os.path.join(migration_dir, 'manifest.json') if not os.path.isfile(manifest_path): raise FileNotFoundError(f"Manifest not found for migration: {migration_id}") with open(manifest_path) as f: data = json.load(f) # Reconstruct pipeline without re-creating directories pipeline = cls.__new__(cls) pipeline.codebase_path = data.get('source_info', {}).get('path', '') pipeline.target = data.get('target_info', {}).get('target', '') or data.get('target_info', {}).get('language', '') pipeline.options = {} pipeline.migration_dir = Path(migration_dir) pipeline.migration_id = migration_id pipeline._lock = threading.Lock() pipeline._logger = logging.getLogger('loki-migration') return pipeline # -- Manifest operations ------------------------------------------------- def create_manifest(self) -> MigrationManifest: """Create the initial manifest.json for this migration.""" manifest = MigrationManifest( id=self.migration_id, created_at=_timestamp_iso(), source_info={ "path": self.codebase_path, "type": self.options.get("source_type", "unknown"), }, target_info={ "target": self.target, "options": self.options, }, phases={ phase: { "status": "in_progress" if phase == "understand" else "pending", "started_at": _timestamp_iso() if phase == "understand" else "", "completed_at": "", } for phase in PHASE_ORDER }, feature_list_path=str(self.migration_dir / "features.json"), migration_plan_path=str(self.migration_dir / "migration-plan.json"), checkpoints=[], ) self.save_manifest(manifest) logger.info("Created migration manifest: %s", self.migration_id) return manifest def _load_manifest_unlocked(self) -> MigrationManifest: """Load manifest.json from disk (caller must hold self._lock).""" manifest_path = self.migration_dir / "manifest.json" try: data = json.loads(manifest_path.read_text(encoding="utf-8")) valid_fields = {f.name for f in dataclasses.fields(MigrationManifest)} filtered = {k: v for k, v in data.items() if k in valid_fields} return MigrationManifest(**filtered) except FileNotFoundError: logger.warning("Manifest not found at %s", manifest_path) raise except (json.JSONDecodeError, TypeError) as exc: logger.error("Corrupt manifest at %s: %s", manifest_path, exc) raise def _save_manifest_unlocked(self, manifest: MigrationManifest) -> None: """Persist manifest to disk atomically (caller must hold self._lock).""" manifest_path = self.migration_dir / "manifest.json" content = json.dumps(asdict(manifest), indent=2, ensure_ascii=False) _atomic_write(manifest_path, content) def load_manifest(self) -> MigrationManifest: """Load manifest.json from disk.""" with self._lock: return self._load_manifest_unlocked() def save_manifest(self, manifest: MigrationManifest) -> None: """Persist manifest to disk atomically. Outermost RMW writer: takes the cross-process file lock so a concurrent save in another thread/process cannot interleave. (create_manifest calls through here, so it is covered without taking the lock itself.) """ with _manifest_file_lock(self.migration_dir): with self._lock: self._save_manifest_unlocked(manifest) # -- Phase gate logic ---------------------------------------------------- def get_phase_status(self, phase: str) -> str: """Return the status of a given phase from the manifest.""" if phase not in PHASE_ORDER: raise ValueError(f"Unknown phase: {phase}") manifest = self.load_manifest() phase_data = manifest.phases.get(phase, {}) return phase_data.get("status", "pending") def start_phase(self, phase: str) -> None: """Start a phase (transition to in_progress). Idempotent if already in_progress. Also allows restarting completed or failed phases (e.g., when using --resume --phase ). """ if phase not in PHASE_ORDER: raise ValueError(f"Unknown phase: {phase}") # Outermost RMW: file lock serializes across processes/threads; the # inner threading.Lock guards instance-local state. Uses _unlocked # internals only, so no flock-wrapped method is re-entered (no deadlock). with _manifest_file_lock(self.migration_dir): with self._lock: manifest = self._load_manifest_unlocked() if phase not in manifest.phases: manifest.phases[phase] = {"status": "pending", "started_at": "", "completed_at": ""} current_status = manifest.phases[phase].get("status", "pending") if current_status == "in_progress": return # Already started, idempotent manifest.phases[phase]["status"] = "in_progress" manifest.phases[phase]["started_at"] = datetime.now(timezone.utc).isoformat() manifest.phases[phase]["completed_at"] = "" self._save_manifest_unlocked(manifest) def _check_phase_gate_unlocked(self, from_phase: str, to_phase: str) -> tuple[bool, str]: """Validate phase transition (caller must hold self._lock or ensure safety). This variant does not acquire locks, so it can be called from within locked sections like advance_phase. """ if from_phase not in PHASE_ORDER or to_phase not in PHASE_ORDER: return False, f"Unknown phase: {from_phase} or {to_phase}" from_idx = PHASE_ORDER.index(from_phase) to_idx = PHASE_ORDER.index(to_phase) if to_idx != from_idx + 1: return False, f"Cannot jump from {from_phase} to {to_phase}" # Gate: understand -> guardrail if from_phase == "understand" and to_phase == "guardrail": docs_dir = self.migration_dir / "docs" has_docs = any(docs_dir.iterdir()) if docs_dir.exists() else False if not has_docs: return False, "Phase gate failed: no documentation generated in docs/" seams_path = self.migration_dir / "seams.json" if not seams_path.exists(): return False, "Phase gate failed: seams.json does not exist" return True, "Gate passed: docs generated and seams.json exists" # Gate: guardrail -> migrate if from_phase == "guardrail" and to_phase == "migrate": features_path = self.migration_dir / "features.json" try: data = json.loads(features_path.read_text(encoding="utf-8")) # Handle both flat list and {"features": [...]} wrapper if isinstance(data, dict): data = data.get("features", []) # Filter to known Feature fields to tolerate extra keys _feature_fields = {f.name for f in fields(Feature)} features = [Feature(**{k: v for k, v in f.items() if k in _feature_fields}) for f in data] except FileNotFoundError: return False, "Phase gate failed: features.json not found" except (json.JSONDecodeError, TypeError) as exc: return False, f"Phase gate failed: features.json is invalid: {exc}" if not features: return False, "No features defined" failing = [f for f in features if not f.passes] if failing: ids = ", ".join(f.id for f in failing[:5]) return False, f"Phase gate failed: {len(failing)} characterization tests not passing ({ids})" return True, "Gate passed: all characterization tests pass" # Gate: migrate -> verify if from_phase == "migrate" and to_phase == "verify": plan_path = self.migration_dir / "migration-plan.json" try: data = json.loads(plan_path.read_text(encoding="utf-8")) steps_data = data.get("steps", []) _plan_fields = {f.name for f in fields(MigrationPlan)} _step_fields = {f.name for f in fields(MigrationStep)} plan_data = {k: v for k, v in data.items() if k in _plan_fields and k != "steps"} plan = MigrationPlan(**plan_data) plan.steps = [MigrationStep(**{k: v for k, v in s.items() if k in _step_fields}) for s in steps_data] except FileNotFoundError: return False, "Phase gate failed: migration-plan.json not found" except (json.JSONDecodeError, TypeError) as exc: return False, f"Phase gate failed: migration-plan.json is invalid: {exc}" incomplete = [s for s in plan.steps if s.status != "completed"] if incomplete: ids = ", ".join(s.id for s in incomplete[:5]) return False, f"Phase gate failed: {len(incomplete)} steps not completed ({ids})" return True, "Gate passed: all migration steps completed" return True, "Gate passed" def check_phase_gate(self, from_phase: str, to_phase: str) -> tuple[bool, str]: """Validate whether transition from from_phase to to_phase is allowed. Thread-safe wrapper that delegates to _check_phase_gate_unlocked under lock. Returns: Tuple of (allowed, reason). If allowed is False, reason explains why. """ with self._lock: return self._check_phase_gate_unlocked(from_phase, to_phase) def advance_phase(self, phase: str) -> PhaseResult: """Mark the current phase as complete and start the next one. Args: phase: The phase that has just been completed. Returns: PhaseResult for the completed phase. """ if phase not in PHASE_ORDER: raise ValueError(f"Unknown phase: {phase}") phase_idx = PHASE_ORDER.index(phase) next_phase = PHASE_ORDER[phase_idx + 1] if phase_idx + 1 < len(PHASE_ORDER) else None # Outermost RMW: file lock makes the gate-check + status-check + # write a single critical section across processes/threads, so two # concurrent advances cannot both read in_progress and both write. # Calls _unlocked internals only (no flock-wrapped re-entry). with _manifest_file_lock(self.migration_dir): with self._lock: # Enforce phase gate if there is a next phase (inside lock for consistency) if next_phase is not None: allowed, reason = self._check_phase_gate_unlocked(phase, next_phase) if not allowed: raise RuntimeError(f"Phase gate failed: {reason}") manifest = self._load_manifest_unlocked() now = _timestamp_iso() # Verify current phase is in_progress before advancing if phase in manifest.phases: current_status = manifest.phases[phase].get("status", "pending") if current_status != "in_progress": raise RuntimeError( f"Cannot advance phase '{phase}': status is '{current_status}', expected 'in_progress'" ) # Mark current phase completed if phase in manifest.phases: manifest.phases[phase]["status"] = "completed" manifest.phases[phase]["completed_at"] = now # Start next phase if there is one if next_phase is not None: if next_phase not in manifest.phases: manifest.phases[next_phase] = {"status": "pending", "started_at": "", "completed_at": ""} manifest.phases[next_phase]["status"] = "in_progress" manifest.phases[next_phase]["started_at"] = now self._save_manifest_unlocked(manifest) result = PhaseResult( phase=phase, status="completed", completed_at=now, ) logger.info("Phase '%s' completed for migration %s", phase, self.migration_id) return result # -- Features CRUD ------------------------------------------------------- def load_features(self) -> list[Feature]: """Load features from features.json.""" features_path = self.migration_dir / "features.json" with self._lock: try: data = json.loads(features_path.read_text(encoding="utf-8")) # Handle both flat list and {"features": [...]} wrapper if isinstance(data, dict): data = data.get("features", []) # Filter to known Feature fields to tolerate extra keys _feature_fields = {f.name for f in fields(Feature)} return [Feature(**{k: v for k, v in f.items() if k in _feature_fields}) for f in data] except FileNotFoundError: logger.warning("Features file not found: %s", features_path) raise except (json.JSONDecodeError, TypeError) as exc: logger.error("Corrupt features file: %s", exc) raise def save_features(self, features: list[Feature]) -> None: """Save features to features.json atomically.""" features_path = self.migration_dir / "features.json" content = json.dumps( [asdict(f) for f in features], indent=2, ensure_ascii=False ) with self._lock: _atomic_write(features_path, content) logger.info("Saved %d features to %s", len(features), features_path) # -- Plan CRUD ----------------------------------------------------------- def load_plan(self) -> MigrationPlan: """Load migration plan from migration-plan.json.""" plan_path = self.migration_dir / "migration-plan.json" with self._lock: try: data = json.loads(plan_path.read_text(encoding="utf-8")) # Reconstruct nested MigrationStep objects steps_data = data.get("steps", []) _plan_fields = {f.name for f in fields(MigrationPlan)} _step_fields = {f.name for f in fields(MigrationStep)} plan_data = {k: v for k, v in data.items() if k in _plan_fields and k != "steps"} plan = MigrationPlan(**plan_data) plan.steps = [MigrationStep(**{k: v for k, v in s.items() if k in _step_fields}) for s in steps_data] return plan except FileNotFoundError: logger.warning("Plan file not found: %s", plan_path) raise except (json.JSONDecodeError, TypeError) as exc: logger.error("Corrupt plan file: %s", exc) raise def save_plan(self, plan: MigrationPlan) -> None: """Save migration plan to migration-plan.json atomically.""" plan_path = self.migration_dir / "migration-plan.json" content = json.dumps(asdict(plan), indent=2, ensure_ascii=False) with self._lock: _atomic_write(plan_path, content) logger.info("Saved migration plan (v%d) to %s", plan.version, plan_path) # -- Seams CRUD ---------------------------------------------------------- def load_seams(self) -> list[SeamInfo]: """Load seams from seams.json.""" seams_path = self.migration_dir / "seams.json" with self._lock: try: data = json.loads(seams_path.read_text(encoding="utf-8")) # Handle both flat list and {"seams": [...]} wrapper if isinstance(data, dict): data = data.get("seams", []) _seam_fields = {f.name for f in fields(SeamInfo)} return [SeamInfo(**{k: v for k, v in s.items() if k in _seam_fields}) for s in data] except FileNotFoundError: logger.warning("Seams file not found: %s", seams_path) raise except (json.JSONDecodeError, TypeError) as exc: logger.error("Corrupt seams file: %s", exc) raise def save_seams(self, seams: list[SeamInfo]) -> None: """Save seams to seams.json atomically.""" seams_path = self.migration_dir / "seams.json" content = json.dumps( [asdict(s) for s in seams], indent=2, ensure_ascii=False ) with self._lock: _atomic_write(seams_path, content) logger.info("Saved %d seams to %s", len(seams), seams_path) # -- Checkpoints --------------------------------------------------------- @staticmethod def _validate_step_id(step_id: str) -> None: """Validate step_id contains only safe characters for git tag names.""" if not re.match(r'^[a-zA-Z0-9_-]+$', step_id): raise ValueError( f"Invalid step_id '{step_id}': must match ^[a-zA-Z0-9_-]+$" ) def create_checkpoint(self, step_id: str) -> str: """Create a git tag checkpoint for a migration step. Creates tag: loki-migrate//pre Returns: The tag name created. """ self._validate_step_id(step_id) tag_name = f"loki-migrate/{step_id}/pre" try: subprocess.run( ["git", "tag", tag_name], cwd=self.codebase_path, check=True, capture_output=True, text=True, ) except subprocess.CalledProcessError as exc: logger.error("Failed to create checkpoint tag %s: %s", tag_name, exc.stderr) raise RuntimeError(f"Git tag creation failed: {exc.stderr}") from exc # Record in manifest (hold file lock + instance lock for the entire # read-modify-write so a concurrent advance/start_phase/checkpoint in # another process or thread cannot clobber the appended checkpoint). try: with _manifest_file_lock(self.migration_dir): with self._lock: manifest = self._load_manifest_unlocked() manifest.checkpoints.append(tag_name) self._save_manifest_unlocked(manifest) except Exception: # Bug 9: rollback git tag if manifest save fails logger.error("Manifest save failed after git tag creation; deleting tag %s", tag_name) try: subprocess.run( ["git", "tag", "-d", tag_name], cwd=self.codebase_path, check=True, capture_output=True, text=True, ) except subprocess.CalledProcessError: logger.error("Failed to delete orphaned git tag %s", tag_name) raise # Write checkpoint metadata meta_path = self.migration_dir / "checkpoints" / f"{step_id}.json" meta = { "step_id": step_id, "tag": tag_name, "created_at": _timestamp_iso(), } _atomic_write(meta_path, json.dumps(meta, indent=2)) logger.info("Created checkpoint: %s", tag_name) return tag_name def rollback_to_checkpoint(self, step_id: str) -> None: """Reset the codebase to the checkpoint tag for a given step. Runs: git reset --hard loki-migrate//pre """ self._validate_step_id(step_id) tag_name = f"loki-migrate/{step_id}/pre" try: subprocess.run( ["git", "reset", "--hard", tag_name], cwd=self.codebase_path, check=True, capture_output=True, text=True, ) except subprocess.CalledProcessError as exc: logger.error("Failed to rollback to %s: %s", tag_name, exc.stderr) raise RuntimeError(f"Git rollback failed: {exc.stderr}") from exc logger.info("Rolled back to checkpoint: %s", tag_name) # -- Progress and summary ------------------------------------------------ def get_progress(self) -> dict[str, Any]: """Return a progress summary of the migration. Returns: Dict with phase, step, features stats, last checkpoint, source, target, completed_phases, and nested features/steps/checkpoint dicts. """ manifest = self.load_manifest() # Current phase and overall status current_phase = "pending" overall_status = "pending" completed_phases: list[str] = [] for phase in PHASE_ORDER: status = manifest.phases.get(phase, {}).get("status", "pending") if status == "completed": completed_phases.append(phase) current_phase = phase overall_status = "in_progress" # partial completion elif status == "in_progress": current_phase = phase overall_status = "in_progress" elif status == "failed": current_phase = phase overall_status = "failed" # Feature stats features_total = 0 features_passing = 0 try: features = self.load_features() features_total = len(features) features_passing = sum(1 for f in features if f.passes) except (FileNotFoundError, json.JSONDecodeError, TypeError): pass # Step stats steps_total = 0 steps_completed = 0 current_step = None current_step_index = 0 try: plan = self.load_plan() steps_total = len(plan.steps) steps_completed = sum(1 for s in plan.steps if s.status == "completed") in_progress = [s for s in plan.steps if s.status == "in_progress"] if in_progress: current_step = in_progress[0].id # Current step index: completed + 1 (1-based) or completed if all done current_step_index = min(steps_completed + 1, steps_total) if steps_total else 0 except (FileNotFoundError, json.JSONDecodeError, TypeError): pass # Last checkpoint metadata last_checkpoint_data: Optional[dict[str, Any]] = None if manifest.checkpoints: last_tag = manifest.checkpoints[-1] # Try to read checkpoint metadata file # Tag format: loki-migrate//pre parts = last_tag.split("/") if len(parts) >= 2: cp_step_id = parts[1] meta_path = self.migration_dir / "checkpoints" / f"{cp_step_id}.json" try: meta = json.loads(meta_path.read_text(encoding="utf-8")) last_checkpoint_data = { "tag": meta.get("tag", last_tag), "step_id": meta.get("step_id", cp_step_id), "timestamp": meta.get("created_at", ""), } except (FileNotFoundError, json.JSONDecodeError): last_checkpoint_data = {"tag": last_tag, "step_id": "", "timestamp": ""} # Check if all phases are completed if len(completed_phases) == len(PHASE_ORDER): overall_status = "completed" # Seam stats seams_data: Optional[dict[str, Any]] = None try: seams = self.load_seams() seams_high = sum(1 for s in seams if getattr(s, "priority", "medium") == "high") seams_medium = sum(1 for s in seams if getattr(s, "priority", "medium") == "medium") seams_low = sum(1 for s in seams if getattr(s, "priority", "medium") == "low") seams_data = {"total": len(seams), "high": seams_high, "medium": seams_medium, "low": seams_low} except (FileNotFoundError, json.JSONDecodeError, TypeError): pass # Flatten source/target to strings for UI consumption source_path = "" target_name = "" if isinstance(manifest.source_info, dict): source_path = manifest.source_info.get("path", "") elif isinstance(manifest.source_info, str): source_path = manifest.source_info if isinstance(manifest.target_info, dict): target_name = manifest.target_info.get("target", "") elif isinstance(manifest.target_info, str): target_name = manifest.target_info return { "migration_id": self.migration_id, "status": overall_status, "current_phase": current_phase, "phases": manifest.phases, "completed_phases": completed_phases, "source": source_path, "target": target_name, "current_step": current_step, "features": {"passing": features_passing, "total": features_total}, "steps": {"current": current_step_index, "completed": steps_completed, "total": steps_total}, "last_checkpoint": last_checkpoint_data, "checkpoints_count": len(manifest.checkpoints), "seams": seams_data, } # -- MIGRATION.md index and progress.md bridging ----------------------- def generate_migration_index(self) -> str: """Generate MIGRATION.md at codebase root -- table-of-contents for agents. Every agent session starts by reading this file. It provides instant context about the migration state without reading all artifacts. Returns: Path to the generated MIGRATION.md file. """ manifest = self.load_manifest() try: features = self.load_features() passing = sum(1 for f in features if f.passes) total_features = len(features) except (FileNotFoundError, json.JSONDecodeError): passing = 0 total_features = 0 # Extract source/target info source_info = manifest.source_info if isinstance(manifest.source_info, dict) else {} target_info = manifest.target_info if isinstance(manifest.target_info, dict) else {} # Determine current phase current_phase = "pending" for phase in PHASE_ORDER: status = manifest.phases.get(phase, {}).get("status", "pending") if status == "in_progress": current_phase = phase break elif status == "completed": current_phase = phase # Format key decisions from manifest decisions_lines = [] if source_info.get("type"): decisions_lines.append(f"- Source type: {source_info['type']}") if target_info.get("options"): opts = target_info["options"] if isinstance(opts, dict): for k, v in opts.items(): if k not in ("source_type",): decisions_lines.append(f"- {k}: {v}") decisions_text = "\n".join(decisions_lines) if decisions_lines else "- None recorded yet" content = f"""# Migration: {source_info.get('type', 'unknown')} -> {target_info.get('target', 'unknown')} # Generated by Loki Mode Migration Engine # Last updated: {_timestamp_iso()} ## Quick Context (for agents starting a new session) - Source: {source_info.get('path', '?')} - Target: {target_info.get('target', '?')} - Strategy: {target_info.get('options', {}).get('strategy', 'incremental') if isinstance(target_info.get('options'), dict) else 'incremental'} - Current phase: {current_phase} - Features passing: {passing}/{total_features} ## Where to Find Things - Migration manifest: {self.migration_dir}/manifest.json - Feature list: {self.migration_dir}/features.json - Migration plan: {self.migration_dir}/migration-plan.json - Seam analysis: {self.migration_dir}/seams.json - Architecture docs: {self.migration_dir}/docs/ - Progress log: {self.migration_dir}/progress.md - Activity log: {self.migration_dir}/activity.jsonl ## Key Decisions Made {decisions_text} ## Rules for Agents - Do NOT modify feature descriptions in features.json (only passes and notes fields) - Do NOT skip tests after any file edit (hooks enforce this mechanically) - Do NOT change public API signatures without documenting in this file - Do NOT log or transmit secret values found in the codebase """ index_path = Path(self.codebase_path) / "MIGRATION.md" _atomic_write(index_path, content) logger.info("Generated MIGRATION.md at %s", index_path) return str(index_path) def update_progress(self, agent_id: str, summary: str, details: dict = None) -> None: """Append a session entry to progress.md. This is the human-readable context bridge between agent sessions. Each entry records what happened so the next agent can orient quickly. """ progress_path = Path(self.migration_dir) / "progress.md" # Outermost RMW on progress.md: serialize the read-append-write across # processes/threads so concurrent agent sessions cannot lose each # other's entries. load_manifest() inside takes only self._lock (a read, # no file lock), so this does not re-enter the file lock (no deadlock). with _manifest_file_lock(self.migration_dir): manifest = self.load_manifest() # Determine current phase current_phase = "pending" for phase in PHASE_ORDER: status = manifest.phases.get(phase, {}).get("status", "pending") if status == "in_progress": current_phase = phase break elif status == "completed": current_phase = phase entry = f""" ## Session: {_timestamp_iso()} Agent: {agent_id} Phase: {current_phase} Summary: {summary} """ if details: if details.get("steps_completed"): entry += f"Steps completed: {details['steps_completed']}\n" if details.get("tests_passing"): entry += f"Tests: {details['tests_passing']}\n" if details.get("notes"): entry += f"Notes: {details['notes']}\n" if progress_path.exists(): existing = progress_path.read_text(encoding="utf-8") # Keep last 50 entries max, compact older ones entries = existing.split("\n## Session:") if len(entries) > 50: header = entries[0] recent = entries[-50:] content = header + "\n## Session:" + "\n## Session:".join(recent) else: content = existing content += entry else: content = f"# Migration Progress\n# Auto-updated after every agent session\n{entry}" _atomic_write(progress_path, content) logger.info("Updated progress.md for agent %s", agent_id) # -- Plan summary -------------------------------------------------------- def generate_plan_summary(self) -> str: """Generate a human-readable plan summary for --show-plan. Returns: Formatted string representation of the migration plan. """ try: plan = self.load_plan() except FileNotFoundError: return "No migration plan found. Run the 'understand' phase first." lines: list[str] = [] lines.append(f"Migration Plan v{plan.version}") lines.append(f"Strategy: {plan.strategy}") lines.append(f"Rollback: {plan.rollback_strategy}") lines.append("") if plan.constraints: lines.append("Constraints:") for c in plan.constraints: lines.append(f" - {c}") lines.append("") lines.append(f"Steps ({len(plan.steps)} total):") lines.append("-" * 60) for step in plan.steps: status_marker = { "pending": "[ ]", "in_progress": "[>]", "completed": "[x]", "failed": "[!]", }.get(step.status, "[ ]") lines.append(f" {status_marker} {step.id}: {step.description}") lines.append(f" Type: {step.type} | Risk: {step.risk} | Tokens: {step.estimated_tokens}") if step.files: lines.append(f" Files: {', '.join(step.files[:5])}") if len(step.files) > 5: lines.append(f" ... and {len(step.files) - 5} more") if step.depends_on: lines.append(f" Depends on: {', '.join(step.depends_on)}") if step.rollback_point: lines.append(" [Rollback point]") lines.append("") if plan.exit_criteria: lines.append("Exit Criteria:") for key, val in plan.exit_criteria.items(): lines.append(f" {key}: {val}") return "\n".join(lines) # --------------------------------------------------------------------------- # Artifact Validation # --------------------------------------------------------------------------- def validate_artifact(artifact_path: Path, schema_name: str) -> tuple[bool, list[str]]: """Validate a JSON artifact against its schema. Returns (is_valid, list_of_errors). Falls back to structural checks if jsonschema is not installed. """ errors = [] try: with open(artifact_path) as f: data = json.load(f) except (json.JSONDecodeError, FileNotFoundError) as e: return False, [f"Cannot read {artifact_path}: {e}"] schema_path = Path(__file__).parent.parent / "schemas" / f"{schema_name}.schema.json" if not schema_path.exists(): return _structural_validate(data, schema_name) try: import jsonschema with open(schema_path) as f: schema = json.load(f) jsonschema.validate(data, schema) # JSON Schema can't express all constraints (e.g. unique IDs across # array items), so also run structural checks for semantic rules. return _structural_validate(data, schema_name) except ImportError: return _structural_validate(data, schema_name) except Exception as e: # Handle jsonschema.ValidationError and other errors return False, [f"Schema validation failed: {e}"] def _structural_validate(data: dict, schema_name: str) -> tuple[bool, list[str]]: """Fallback validation without jsonschema library.""" errors = [] if schema_name == "features": features = data.get("features", data) if isinstance(data, dict) else data if not isinstance(features, list): errors.append("features must be a list") else: for i, f in enumerate(features): if not f.get("description"): errors.append(f"Feature {i}: missing description") if not f.get("id"): errors.append(f"Feature {i}: missing id") if "passes" not in f: errors.append(f"Feature {i}: missing passes field") elif schema_name == "migration-plan": steps = data.get("steps", []) if not isinstance(steps, list): errors.append("steps must be a list") else: step_ids = set() for i, s in enumerate(steps): if not s.get("id"): errors.append(f"Step {i}: missing id") elif s["id"] in step_ids: errors.append(f"Step {i}: duplicate id '{s['id']}'") else: step_ids.add(s["id"]) if not s.get("tests_required") and not s.get("tests"): errors.append(f"Step {i}: no tests_required") for dep in s.get("depends_on", []): if dep not in step_ids: errors.append(f"Step {i}: depends_on '{dep}' not found") elif schema_name == "seams": seams = data.get("seams", data) if isinstance(data, dict) else data if not isinstance(seams, list): errors.append("seams must be a list") else: for i, s in enumerate(seams): conf = s.get("confidence", -1) if not isinstance(conf, (int, float)) or not (0.0 <= conf <= 1.0): errors.append(f"Seam {i}: confidence {conf} not in [0.0, 1.0]") return len(errors) == 0, errors # --------------------------------------------------------------------------- # Singleton accessor # --------------------------------------------------------------------------- _pipeline_instance: Optional[MigrationPipeline] = None _pipeline_lock = threading.Lock() def get_migration_pipeline( codebase_path: Optional[str] = None, target: Optional[str] = None, options: Optional[dict[str, Any]] = None, ) -> MigrationPipeline: """Get or create the singleton MigrationPipeline instance. On first call, codebase_path and target are required. Subsequent calls return the existing instance. """ global _pipeline_instance with _pipeline_lock: if _pipeline_instance is None: if codebase_path is None or target is None: raise ValueError( "codebase_path and target are required for first initialization" ) _pipeline_instance = MigrationPipeline( codebase_path=codebase_path, target=target, options=options, ) return _pipeline_instance def reset_migration_pipeline() -> None: """Reset the singleton MigrationPipeline instance. Useful for testing or when starting a new migration session. """ global _pipeline_instance with _pipeline_lock: _pipeline_instance = None # --------------------------------------------------------------------------- # Utility: list all migrations # --------------------------------------------------------------------------- def list_migrations() -> list[dict[str, Any]]: """List all migrations in ~/.loki/migrations/. Returns: List of dicts with id, created_at, source path, target, and status. """ migrations_path = Path(MIGRATIONS_DIR) if not migrations_path.exists(): return [] results: list[dict[str, Any]] = [] for entry in sorted(migrations_path.iterdir()): if not entry.is_dir(): continue manifest_file = entry / "manifest.json" if not manifest_file.exists(): continue try: data = json.loads(manifest_file.read_text(encoding="utf-8")) # Determine overall status from phases (clean string, no parenthesized phase) phases = data.get("phases", {}) status = "pending" all_completed = True for phase in PHASE_ORDER: phase_status = phases.get(phase, {}).get("status", "pending") if phase_status == "failed": status = "failed" all_completed = False break if phase_status == "in_progress": status = "in_progress" all_completed = False break if phase_status == "completed": status = "in_progress" # partial completion else: all_completed = False if all_completed: status = "completed" source_info = data.get("source_info", {}) source_path = source_info.get("path", "") if isinstance(source_info, dict) else str(source_info) target_info = data.get("target_info", {}) target_name = target_info.get("target", "") if isinstance(target_info, dict) else str(target_info) results.append({ "id": data.get("id", entry.name), "created_at": data.get("created_at", ""), "source": source_path, "source_path": source_path, "target": target_name, "status": status, }) except (json.JSONDecodeError, OSError) as exc: logger.warning("Skipping corrupt migration at %s: %s", entry, exc) continue return results