"""muse code migrate — Replay a Git commit graph into a Muse repository. Converts an existing Git repository into a Muse repository by replaying every requested branch oldest-first, preserving original author names, e-mail addresses, and committer timestamps verbatim in every CommitRecord. Examples -------- :: # Migrate the git repo in the current directory (main branch only) muse code migrate # Migrate into a separate target directory muse code migrate /path/to/git-repo --target /path/to/muse-repo # Replay all local branches muse code migrate --all # Replay specific branches in order muse code migrate --branch main dev feat/x # Preview without writing anything muse code migrate --all --dry-run # Incremental: only replay commits after a known git SHA muse code migrate --from-ref abc123 # Add custom exclusion patterns on top of the defaults muse code migrate --exclude "dist/" "*.lock" # Emit structured NDJSON (useful in agent pipelines — streams progress events) muse code migrate --json Strategy -------- 1. Walk requested branches oldest-first. ``main`` is always replayed first so other branches can branch from the correct Muse ancestor. 2. Merge commits (more than one parent) are skipped — they carry no unique file-state delta; the Muse DAG is reconstructed faithfully through the parent chain on each branch. 3. For each commit, ``git diff-tree --root -r --raw`` fetches only the *delta* (added / modified / deleted / renamed files) rather than the full tree. A running manifest is maintained in memory and updated incrementally — unchanged files are never re-read or re-hashed. 4. All blob content is streamed through a single long-running ``git cat-file --batch`` process (one per branch) so there is no per-file subprocess overhead. 5. If ``.muse/repo.json`` does not exist, ``muse init --domain code`` is run automatically before migration begins. """ from __future__ import annotations import argparse import dataclasses import datetime import hashlib import json import logging import os import pathlib import subprocess import sys import time from muse.core.object_store import write_object from muse.core.reflog import append_reflog from muse.core.snapshot import compute_snapshot_id, compute_commit_id from muse.core.store import ( CommitRecord, SnapshotRecord, write_commit, write_head_branch, write_snapshot, ) logger = logging.getLogger(__name__) # Type alias — maps relative POSIX path → muse object id (sha256 hex) Manifest = dict[str, str] # --------------------------------------------------------------------------- # Default exclusion rules (mirror .museignore + hidden-dir walk logic) # --------------------------------------------------------------------------- _DEFAULT_EXCLUDE_PREFIXES: tuple[str, ...] = ( ".git/", ".muse/", ".venv/", ".tox/", ".mypy_cache/", ".pytest_cache/", ".hypothesis/", "artifacts/", "__pycache__/", "node_modules/", ".cache/", ) _DEFAULT_EXCLUDE_SUFFIXES: tuple[str, ...] = ( ".pyc", ".pyo", ".egg-info", ".swp", ".swo", ".tmp", "Thumbs.db", ".DS_Store", ) def _should_exclude( rel_path: str, extra_prefixes: tuple[str, ...], extra_suffixes: tuple[str, ...], ) -> bool: all_prefixes = _DEFAULT_EXCLUDE_PREFIXES + extra_prefixes all_suffixes = _DEFAULT_EXCLUDE_SUFFIXES + extra_suffixes for prefix in all_prefixes: if rel_path.startswith(prefix) or rel_path == prefix.rstrip("/"): return True for suffix in all_suffixes: if rel_path.endswith(suffix): return True return False # --------------------------------------------------------------------------- # Delta entry — one changed file from git diff-tree # --------------------------------------------------------------------------- @dataclasses.dataclass(slots=True) class _Delta: status: str # A add, M modify, D delete, R rename, C copy, T type-change blob_sha: str # git blob SHA of the new content (empty string for deletes) path: str # destination path old_path: str # source path for renames/copies, else same as path # --------------------------------------------------------------------------- # git cat-file --batch — long-running process for streaming blob reads # --------------------------------------------------------------------------- class _CatFile: """Wraps a ``git cat-file --batch`` subprocess for efficient blob reads. Maintains one long-running process for the duration of a branch replay, eliminating the per-file subprocess spawn overhead that dominates small repos and becomes catastrophic for large ones. """ def __init__(self, git_root: pathlib.Path) -> None: proc = subprocess.Popen( ["git", "cat-file", "--batch"], cwd=git_root, stdin=subprocess.PIPE, stdout=subprocess.PIPE, ) if proc.stdin is None or proc.stdout is None: proc.kill() raise RuntimeError("git cat-file --batch failed to open stdin/stdout pipes") self._proc = proc self._stdin = proc.stdin self._stdout = proc.stdout def read(self, sha: str) -> bytes: """Return the raw blob content for *sha*; empty bytes if missing or deleted.""" self._stdin.write(f"{sha}\n".encode()) self._stdin.flush() header = self._stdout.readline().decode(errors="replace") parts = header.split() if len(parts) < 3 or parts[1] == "missing": return b"" size = int(parts[2]) content = self._stdout.read(size) self._stdout.read(1) # consume trailing newline return content def close(self) -> None: """Close the subprocess stdin and wait for it to exit cleanly.""" self._stdin.close() self._proc.wait() def __enter__(self) -> "_CatFile": return self def __exit__(self, *_: object) -> None: self.close() # --------------------------------------------------------------------------- # git helpers # --------------------------------------------------------------------------- def _git(repo_root: pathlib.Path, *args: str) -> str: result = subprocess.run( ["git", *args], cwd=repo_root, capture_output=True, text=True, check=True, ) return result.stdout.strip() def _git_local_branches(repo_root: pathlib.Path) -> list[str]: raw = _git(repo_root, "branch", "--format=%(refname:short)") return [b.strip() for b in raw.splitlines() if b.strip()] def _git_is_repo(path: pathlib.Path) -> bool: try: _git(path, "rev-parse", "--git-dir") return True except (subprocess.CalledProcessError, FileNotFoundError): return False # ASCII control characters as field/record separators — safe in commit messages. _FIELD_SEP = "\x1f" # Unit Separator _RECORD_SEP = "\x1e" # Record Separator def _batch_commit_log( repo_root: pathlib.Path, branch: str, exclude_branches: list[str] | None = None, from_ref: str | None = None, ) -> list[tuple[str, dict[str, str]]]: """Return ``[(sha, meta_dict), ...]`` oldest-first in a single git log call. Streams git log output in 64 KiB chunks and parses records on the fly — the raw output is never fully buffered in memory. Critical for repos with hundreds of thousands of commits where a single ``git log`` call can produce gigabytes of output. """ fmt = ( f"%H{_FIELD_SEP}%an{_FIELD_SEP}%ae{_FIELD_SEP}" f"%at{_FIELD_SEP}%P{_FIELD_SEP}%B{_RECORD_SEP}" ) cmd = ["log", "--topo-order", "--reverse", f"--format={fmt}"] if exclude_branches: cmd.append(branch) for excl in exclude_branches: cmd.append(f"^{excl}") elif from_ref: cmd.append(f"{from_ref}..{branch}") else: cmd.append(branch) proc = subprocess.Popen( ["git", *cmd], cwd=repo_root, stdout=subprocess.PIPE, text=True, ) if proc.stdout is None: proc.kill() raise RuntimeError("git log failed to open stdout pipe") results: list[tuple[str, dict[str, str]]] = [] buf = "" _chunk = 65536 try: while True: chunk = proc.stdout.read(_chunk) if not chunk: break buf += chunk while _RECORD_SEP in buf: record, buf = buf.split(_RECORD_SEP, 1) record = record.strip() if not record: continue parts = record.split(_FIELD_SEP, 5) if len(parts) < 6: continue sha, name, email, ts, parents, message = parts results.append((sha.strip(), { "name": name.strip(), "email": email.strip(), "ts": ts.strip(), "parents": parents.strip(), "message": message.strip(), })) # Flush any trailing record not followed by a separator. record = buf.strip() if record: parts = record.split(_FIELD_SEP, 5) if len(parts) >= 6: sha, name, email, ts, parents, message = parts results.append((sha.strip(), { "name": name.strip(), "email": email.strip(), "ts": ts.strip(), "parents": parents.strip(), "message": message.strip(), })) finally: proc.stdout.close() proc.wait() return results def _batch_diff_tree( repo_root: pathlib.Path, shas: list[str], ) -> dict[str, list[_Delta]]: """Return ``{sha: [delta, ...]}`` for all *shas* in one ``git diff-tree --stdin`` call. Streams stdout line by line — the entire diff-tree output is never buffered in memory at once. For a repository with 1 M commits and ~10 changed files per commit, ``communicate()`` would buffer ~2 GB; streaming keeps memory proportional to the largest single-commit diff, not the entire history. ``--root`` ensures the first (root) commit is compared against the empty tree so all its files appear as additions. """ if not shas: return {} proc = subprocess.Popen( ["git", "diff-tree", "--stdin", "--root", "-r", "--raw", "--no-abbrev"], cwd=repo_root, stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True, ) if proc.stdin is None or proc.stdout is None: proc.kill() raise RuntimeError("git diff-tree failed to open stdin/stdout pipes") # Write all SHAs to stdin then close it so git starts processing. proc.stdin.write("\n".join(shas) + "\n") proc.stdin.close() result: dict[str, list[_Delta]] = {sha: [] for sha in shas} current_sha: str | None = None sha_set = set(shas) try: for line in proc.stdout: line = line.rstrip("\n") if not line: continue if not line.startswith(":"): # Commit SHA header line — first token is the SHA. candidate = line.split()[0] if line.split() else "" if candidate in sha_set: current_sha = candidate continue if current_sha is None: continue meta_part, _tab, path_part = line.partition("\t") fields = meta_part.split() if len(fields) < 5: continue new_sha = fields[3] status = fields[4][0] if status in ("R", "C"): old_path, _, new_path = path_part.partition("\t") result[current_sha].append( _Delta(status=status, blob_sha=new_sha, path=new_path, old_path=old_path) ) else: result[current_sha].append( _Delta(status=status, blob_sha=new_sha, path=path_part, old_path=path_part) ) finally: proc.stdout.close() proc.wait() return result # --------------------------------------------------------------------------- # Manifest helpers # --------------------------------------------------------------------------- def _manifest_directories(manifest: Manifest) -> list[str]: """Derive the sorted set of directory paths from the manifest keys. Git does not track empty directories — every directory in a migrated snapshot is implied by its file entries. This produces a consistent set that matches what ``walk_workdir_with_dirs`` would return on a filesystem that mirrors the manifest. """ dirs: set[str] = set() for path in manifest: p = pathlib.PurePosixPath(path) for parent in p.parents: s = str(parent) if s and s != ".": dirs.add(s) return sorted(dirs) # --------------------------------------------------------------------------- # Muse ref helpers # --------------------------------------------------------------------------- def _refs_dir(muse_root: pathlib.Path) -> pathlib.Path: return muse_root / ".muse" / "refs" / "heads" def _set_branch_head(muse_root: pathlib.Path, branch: str, commit_id: str) -> None: ref_path = _refs_dir(muse_root) / branch ref_path.parent.mkdir(parents=True, exist_ok=True) ref_path.write_text(commit_id + "\n") def _get_branch_head(muse_root: pathlib.Path, branch: str) -> str | None: ref_path = _refs_dir(muse_root) / branch if not ref_path.exists(): return None return ref_path.read_text().strip() or None def _ensure_branch_exists(muse_root: pathlib.Path, branch: str) -> None: _refs_dir(muse_root).mkdir(parents=True, exist_ok=True) ref_path = _refs_dir(muse_root) / branch if not ref_path.exists(): ref_path.write_text("") def _load_repo_id(muse_root: pathlib.Path) -> str: repo_json = muse_root / ".muse" / "repo.json" data: dict[str, str] = json.loads(repo_json.read_text()) return data["repo_id"] # --------------------------------------------------------------------------- # Core replay logic — incremental manifest, no temp workdir # --------------------------------------------------------------------------- def _replay_commit( muse_root: pathlib.Path, manifest: Manifest, muse_branch: str, parent_muse_id: str | None, meta: dict[str, str], repo_id: str, dry_run: bool, ) -> str: """Write one Muse commit from the current *manifest* state. The manifest has already been updated by the caller (added/deleted files applied). This function computes the snapshot ID, writes the snapshot and commit records, and returns the new Muse commit ID. """ directories = _manifest_directories(manifest) snapshot_id = compute_snapshot_id(manifest, directories=directories) committed_at = datetime.datetime.fromtimestamp( int(meta["ts"]), tz=datetime.timezone.utc ) author = f"{meta['name']} <{meta['email']}>" message = meta["message"] or "no message" committed_at_iso = committed_at.isoformat() parent_ids = [parent_muse_id] if parent_muse_id else [] commit_id = compute_commit_id( parent_ids=parent_ids, snapshot_id=snapshot_id, message=message, committed_at_iso=committed_at_iso, ) if dry_run: return commit_id write_snapshot( muse_root, SnapshotRecord( snapshot_id=snapshot_id, manifest=dict(manifest), directories=directories, ), ) write_commit( muse_root, CommitRecord( commit_id=commit_id, repo_id=repo_id, branch=muse_branch, snapshot_id=snapshot_id, message=message, committed_at=committed_at, parent_commit_id=parent_muse_id, author=author, ), ) _set_branch_head(muse_root, muse_branch, commit_id) append_reflog( muse_root, muse_branch, old_id=parent_muse_id, new_id=commit_id, author=author, operation=f"migrate: {message[:60]}", ) return commit_id def _replay_branch( git_root: pathlib.Path, muse_root: pathlib.Path, git_shas: list[str], muse_branch: str, start_parent_muse_id: str | None, repo_id: str, dry_run: bool, verbose: bool, extra_prefixes: tuple[str, ...], extra_suffixes: tuple[str, ...], as_json: bool, initial_manifest: Manifest, all_meta: dict[str, dict[str, str]], ) -> tuple[dict[str, str], Manifest]: """Replay *git_shas* (oldest-first) onto *muse_branch*. Uses three batched git processes for the entire branch (not one per commit): 1. ``git diff-tree --stdin`` — all per-file deltas in one call. 2. ``git cat-file --batch`` — all blob content streamed through one process. Commit metadata is pre-loaded in *all_meta* (from a single ``git log`` call in the caller). The manifest is maintained incrementally: only added/modified/deleted files are touched per commit, so unchanged files are never re-read or re-hashed. Returns ``(git_sha → muse_commit_id mapping, final manifest state)``. """ _ensure_branch_exists(muse_root, muse_branch) git_to_muse: dict[str, str] = {} parent_muse_id = start_parent_muse_id manifest: Manifest = dict(initial_manifest) total = len(git_shas) # Batch all delta reads in a single git diff-tree --stdin call. all_deltas = _batch_diff_tree(git_root, git_shas) _PROGRESS_EVERY = 100 # emit a JSON progress event every N commits with _CatFile(git_root) as cat: for i, sha in enumerate(git_shas, 1): meta = all_meta[sha] if as_json: if i == 1 or i == total or i % _PROGRESS_EVERY == 0: print(json.dumps({ "event": "progress", "branch": muse_branch, "committed": i, "total": total, }), flush=True) elif verbose or i % 50 == 0 or i == 1 or i == total: logger.info( "[%s] %d/%d git:%s %s", muse_branch, i, total, sha[:12], repr(meta["message"][:60]), ) if not dry_run: for delta in all_deltas.get(sha, []): if _should_exclude(delta.path, extra_prefixes, extra_suffixes): manifest.pop(delta.path, None) continue if delta.status == "D": manifest.pop(delta.path, None) continue if delta.status in ("R", "C"): manifest.pop(delta.old_path, None) null40 = "0" * 40 if delta.blob_sha and delta.blob_sha != null40: content = cat.read(delta.blob_sha) if content: oid = hashlib.sha256(content).hexdigest() write_object(muse_root, oid, content) manifest[delta.path] = oid muse_id = _replay_commit( muse_root=muse_root, manifest=manifest, muse_branch=muse_branch, parent_muse_id=parent_muse_id, meta=meta, repo_id=repo_id, dry_run=dry_run, ) git_to_muse[sha] = muse_id parent_muse_id = muse_id return git_to_muse, manifest # --------------------------------------------------------------------------- # CLI wiring # --------------------------------------------------------------------------- def register( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register the ``migrate`` subcommand under ``muse code``.""" parser = subparsers.add_parser( "migrate", help="Replay a Git repository's commit history into a Muse repository.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "source", nargs="?", type=pathlib.Path, default=None, metavar="SOURCE", help=( "Path to the Git repository to migrate from. " "Defaults to the current working directory." ), ) parser.add_argument( "--target", type=pathlib.Path, default=None, metavar="PATH", dest="target", help=( "Path to write the Muse repository into. " "Defaults to SOURCE (migrate in-place)." ), ) branch_group = parser.add_mutually_exclusive_group() branch_group.add_argument( "--branch", nargs="+", default=None, metavar="BRANCH", dest="branches", help=( "Git branch(es) to replay in the given order (default: main). " "The first branch is treated as primary and replayed in full; " "subsequent branches replay only commits not reachable from any " "previously replayed branch." ), ) branch_group.add_argument( "--all", action="store_true", dest="all_branches", help="Replay every local branch (main first, then the rest alphabetically).", ) parser.add_argument( "--from-ref", default=None, metavar="SHA", dest="from_ref", help=( "Only replay commits newer than this Git SHA. " "Useful for incremental re-runs after initial migration." ), ) parser.add_argument( "--exclude", nargs="+", default=[], metavar="PATTERN", dest="excludes", help=( "Additional path prefixes (ending with '/') or suffixes (starting " "with '.') to exclude, on top of the built-in defaults." ), ) parser.add_argument( "--no-init", action="store_true", dest="no_init", help="Skip auto-initialisation when the target already has a Muse repo.", ) parser.add_argument( "--dry-run", action="store_true", dest="dry_run", help="Log what would happen without writing anything to disk.", ) parser.add_argument( "--verbose", "-v", action="store_true", dest="verbose", help="Log every commit (default: log first, last, and every 50th).", ) parser.add_argument( "--json", action="store_true", dest="as_json", help=( "Emit NDJSON progress events during migration and a final 'done' " "summary. Each line is a JSON object with an 'event' field: " "'branch_start', 'progress', 'branch_done', or 'done'." ), ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Entry point for ``muse code migrate``.""" logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") source: pathlib.Path = (args.source or pathlib.Path.cwd()).resolve() target: pathlib.Path = (args.target or source).resolve() dry_run: bool = args.dry_run verbose: bool = args.verbose as_json: bool = args.as_json no_init: bool = args.no_init from_ref: str | None = args.from_ref excludes: list[str] = args.excludes extra_prefixes: tuple[str, ...] = tuple(e for e in excludes if e.endswith("/")) extra_suffixes: tuple[str, ...] = tuple(e for e in excludes if not e.endswith("/")) if not _git_is_repo(source): logger.error("❌ %s is not a Git repository.", source) sys.exit(1) if args.all_branches: all_branches = _git_local_branches(source) primaries = [b for b in all_branches if b == "main"] others = sorted(b for b in all_branches if b != "main") requested: list[str] = primaries + others else: requested = args.branches or ["main"] if not requested: logger.error("❌ No branches to replay.") sys.exit(1) if not as_json: logger.info("━━━ muse code migrate ━━━") logger.info(" source : %s", source) logger.info(" target : %s", target) logger.info(" branches: %s", requested) if dry_run: logger.info(" mode : DRY RUN — nothing will be written") muse_json = target / ".muse" / "repo.json" if not muse_json.exists() and not no_init: if not as_json: logger.info("No .muse/repo.json — running 'muse init --domain code' …") if not dry_run: target.mkdir(parents=True, exist_ok=True) result = subprocess.run( ["muse", "init", "--domain", "code"], cwd=target, capture_output=True, text=True, ) if result.returncode != 0: logger.error("❌ muse init failed:\n%s", result.stderr) sys.exit(1) if not as_json: logger.info("✅ muse init succeeded") repo_id = _load_repo_id(target) if muse_json.exists() else "dry-run-placeholder" started_at = time.monotonic() all_git_to_muse: dict[str, str] = {} replayed_branches: list[str] = [] branch_stats: list[dict[str, object]] = [] # Carry the final manifest of the primary branch so feature branches # start from the right tree state instead of from empty. primary_final_manifest: Manifest = {} for branch in requested: if not as_json: logger.info("━━━ Replaying branch: %s ━━━", branch) exclude_refs = replayed_branches if replayed_branches else None # One git log call: all SHA + metadata for the branch, oldest first. commit_log = _batch_commit_log( source, branch, exclude_branches=exclude_refs, from_ref=from_ref if not exclude_refs else None, ) # Split non-merge from merge commits using the pre-loaded parent field. non_merge: list[tuple[str, dict[str, str]]] = [] merge_count = 0 for sha, meta in commit_log: parents = meta.get("parents", "").split() if len(parents) > 1: merge_count += 1 else: non_merge.append((sha, meta)) git_shas = [sha for sha, _ in non_merge] all_meta: dict[str, dict[str, str]] = {sha: meta for sha, meta in non_merge} if as_json: print(json.dumps({ "event": "branch_start", "branch": branch, "total_commits": len(git_shas), "merge_commits_skipped": merge_count, }), flush=True) else: logger.info( " %d commits unique to %s (%d merge commits skipped)", len(git_shas), branch, merge_count, ) if not git_shas: if not as_json: logger.info(" %s has no unique commits — skipping", branch) replayed_branches.append(branch) stat: dict[str, object] = { "branch": branch, "commits_written": 0, "merge_commits_skipped": merge_count, "skipped": True, } branch_stats.append(stat) if as_json: print(json.dumps({"event": "branch_done", **stat}), flush=True) continue # Find the Muse parent commit to branch from. start_parent_muse_id: str | None = None oldest_sha = git_shas[0] oldest_parents = all_meta[oldest_sha].get("parents", "").split() for gp in oldest_parents: if gp in all_git_to_muse: start_parent_muse_id = all_git_to_muse[gp] break if start_parent_muse_id is None and replayed_branches: start_parent_muse_id = _get_branch_head(target, replayed_branches[0]) # Secondary branches start from the primary's final tree state so the # incremental diff-tree approach produces correct manifests. initial_manifest: Manifest = ( {} if not replayed_branches else dict(primary_final_manifest) ) write_head_branch(target, branch) mapping, final_manifest = _replay_branch( git_root=source, muse_root=target, git_shas=git_shas, muse_branch=branch, start_parent_muse_id=start_parent_muse_id, repo_id=repo_id, dry_run=dry_run, verbose=verbose, extra_prefixes=extra_prefixes, extra_suffixes=extra_suffixes, as_json=as_json, initial_manifest=initial_manifest, all_meta=all_meta, ) all_git_to_muse.update(mapping) replayed_branches.append(branch) if not replayed_branches[1:]: # first branch = primary primary_final_manifest = final_manifest branch_stat: dict[str, object] = { "branch": branch, "commits_written": len(mapping), "merge_commits_skipped": merge_count, "skipped": False, } branch_stats.append(branch_stat) if as_json: print(json.dumps({"event": "branch_done", **branch_stat}), flush=True) else: logger.info("✅ %s: %d commits written", branch, len(mapping)) if not dry_run and replayed_branches: write_head_branch(target, replayed_branches[0]) elapsed = time.monotonic() - started_at total_written = len(all_git_to_muse) if as_json: print(json.dumps({ "event": "done", "source": str(source), "target": str(target), "dry_run": dry_run, "branches": branch_stats, "total_commits_written": total_written, "elapsed_seconds": round(elapsed, 2), }), flush=True) else: logger.info( "━━━ Done ━━━ %d commits written across %d branch(es) in %.1fs", total_written, len(replayed_branches), elapsed, ) if dry_run: logger.info("(dry-run — nothing was written)")