checkout.py
python
| 1 | """muse checkout — create and switch local branches, update .muse/HEAD. |
| 2 | |
| 3 | Behavior |
| 4 | -------- |
| 5 | * ``muse checkout <branch>`` — switch to an existing branch. |
| 6 | * ``muse checkout -b <branch>`` — create a new branch forked from the |
| 7 | current HEAD commit and switch to it. |
| 8 | * Dirty working-tree guard: if ``muse-work/`` differs from the last |
| 9 | committed snapshot the command exits ``1`` with a message, unless |
| 10 | ``--force`` / ``-f`` is supplied. |
| 11 | |
| 12 | Dirty detection |
| 13 | --------------- |
| 14 | The working tree is considered dirty when: |
| 15 | |
| 16 | 1. ``muse-work/`` exists **and** contains at least one file, **and** |
| 17 | 2. Its computed ``snapshot_id`` (``sha256`` of the sorted |
| 18 | ``path:object_id`` pairs) differs from the ``snapshot_id`` of the |
| 19 | most recent commit on the *current* branch. |
| 20 | |
| 21 | If the branch has no commits yet (empty branch) the tree is never |
| 22 | considered dirty — there is nothing to diverge from. |
| 23 | |
| 24 | Branch state |
| 25 | ------------ |
| 26 | Branches are tracked purely on the local filesystem under |
| 27 | ``.muse/refs/heads/<name>``. A DB-level branch table is deferred to |
| 28 | the ``muse merge`` iteration (issue #35) when multi-branch DAG queries |
| 29 | will require it. |
| 30 | """ |
| 31 | from __future__ import annotations |
| 32 | |
| 33 | import asyncio |
| 34 | import json |
| 35 | import logging |
| 36 | import pathlib |
| 37 | import re |
| 38 | |
| 39 | import typer |
| 40 | from sqlalchemy.ext.asyncio import AsyncSession |
| 41 | |
| 42 | from maestro.muse_cli._repo import require_repo |
| 43 | from maestro.muse_cli.db import get_head_snapshot_id, open_session |
| 44 | from maestro.muse_cli.errors import ExitCode |
| 45 | from maestro.muse_cli.snapshot import build_snapshot_manifest, compute_snapshot_id |
| 46 | |
| 47 | logger = logging.getLogger(__name__) |
| 48 | |
| 49 | # Branch names follow the same rules as Git: no spaces, no control chars, |
| 50 | # no leading dots, no double dots, no trailing slash or dot. |
| 51 | _BRANCH_RE = re.compile(r"^[a-zA-Z0-9._\-/]+$") |
| 52 | |
| 53 | |
| 54 | # --------------------------------------------------------------------------- |
| 55 | # Helpers |
| 56 | # --------------------------------------------------------------------------- |
| 57 | |
| 58 | |
| 59 | def _validate_branch_name(name: str) -> None: |
| 60 | """Exit ``1`` if *name* is not a valid branch identifier.""" |
| 61 | if not _BRANCH_RE.match(name) or ".." in name or name.startswith("."): |
| 62 | typer.echo( |
| 63 | f"❌ Invalid branch name '{name}'. " |
| 64 | "Use letters, digits, hyphens, underscores, dots, or forward slashes." |
| 65 | ) |
| 66 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 67 | |
| 68 | |
| 69 | async def _is_dirty( |
| 70 | session: AsyncSession, |
| 71 | root: pathlib.Path, |
| 72 | repo_id: str, |
| 73 | branch: str, |
| 74 | ) -> bool: |
| 75 | """Return ``True`` if ``muse-work/`` has uncommitted changes. |
| 76 | |
| 77 | Compares the on-disk snapshot against the last committed snapshot on |
| 78 | *branch*. Returns ``False`` when the branch has no commits yet. |
| 79 | """ |
| 80 | workdir = root / "muse-work" |
| 81 | if not workdir.exists(): |
| 82 | return False |
| 83 | manifest = build_snapshot_manifest(workdir) |
| 84 | if not manifest: |
| 85 | return False |
| 86 | current_sid = compute_snapshot_id(manifest) |
| 87 | last_sid = await get_head_snapshot_id(session, repo_id, branch) |
| 88 | if last_sid is None: |
| 89 | return False # no commits on this branch — nothing to dirty against |
| 90 | return current_sid != last_sid |
| 91 | |
| 92 | |
| 93 | # --------------------------------------------------------------------------- |
| 94 | # Testable async core |
| 95 | # --------------------------------------------------------------------------- |
| 96 | |
| 97 | |
| 98 | async def _checkout_async( |
| 99 | *, |
| 100 | branch_name: str, |
| 101 | create: bool, |
| 102 | force: bool, |
| 103 | root: pathlib.Path, |
| 104 | session: AsyncSession, |
| 105 | ) -> None: |
| 106 | """Core checkout logic, fully injectable for tests. |
| 107 | |
| 108 | Raises ``typer.Exit`` on every terminal condition (success or error) |
| 109 | so callers do not need to distinguish between return paths. |
| 110 | """ |
| 111 | _validate_branch_name(branch_name) |
| 112 | |
| 113 | muse_dir = root / ".muse" |
| 114 | repo_data: dict[str, str] = json.loads((muse_dir / "repo.json").read_text()) |
| 115 | repo_id = repo_data["repo_id"] |
| 116 | |
| 117 | head_ref = (muse_dir / "HEAD").read_text().strip() # "refs/heads/main" |
| 118 | current_branch = head_ref.rsplit("/", 1)[-1] # "main" |
| 119 | |
| 120 | ref_path = muse_dir / "refs" / "heads" / branch_name |
| 121 | |
| 122 | if create: |
| 123 | # ── muse checkout -b <branch> ──────────────────────────────────── |
| 124 | if ref_path.exists() and ref_path.read_text().strip(): |
| 125 | typer.echo(f"❌ Branch '{branch_name}' already exists.") |
| 126 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 127 | |
| 128 | if not force and await _is_dirty(session, root, repo_id, current_branch): |
| 129 | typer.echo( |
| 130 | "❌ Uncommitted changes in muse-work/. " |
| 131 | "Commit your changes or use --force to override." |
| 132 | ) |
| 133 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 134 | |
| 135 | # Fork from current HEAD commit |
| 136 | current_ref = muse_dir / "refs" / "heads" / current_branch |
| 137 | current_commit_id = ( |
| 138 | current_ref.read_text().strip() if current_ref.exists() else "" |
| 139 | ) |
| 140 | ref_path.parent.mkdir(parents=True, exist_ok=True) |
| 141 | ref_path.write_text(current_commit_id) |
| 142 | (muse_dir / "HEAD").write_text(f"refs/heads/{branch_name}") |
| 143 | |
| 144 | origin = f" (from commit {current_commit_id[:8]})" if current_commit_id else "" |
| 145 | typer.echo(f"✅ Switched to a new branch '{branch_name}'{origin}") |
| 146 | logger.info("✅ Created and switched to branch %r at %s", branch_name, current_commit_id[:8] if current_commit_id else "empty") |
| 147 | |
| 148 | else: |
| 149 | # ── muse checkout <branch> ─────────────────────────────────────── |
| 150 | if not ref_path.exists(): |
| 151 | typer.echo( |
| 152 | f"❌ Branch '{branch_name}' does not exist. " |
| 153 | "Use -b to create it." |
| 154 | ) |
| 155 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 156 | |
| 157 | if branch_name == current_branch: |
| 158 | typer.echo(f"Already on '{branch_name}'") |
| 159 | raise typer.Exit(code=ExitCode.SUCCESS) |
| 160 | |
| 161 | if not force and await _is_dirty(session, root, repo_id, current_branch): |
| 162 | typer.echo( |
| 163 | "❌ Uncommitted changes in muse-work/. " |
| 164 | "Commit your changes or use --force to override." |
| 165 | ) |
| 166 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 167 | |
| 168 | (muse_dir / "HEAD").write_text(f"refs/heads/{branch_name}") |
| 169 | typer.echo(f"✅ Switched to branch '{branch_name}'") |
| 170 | logger.info("✅ Switched to branch %r", branch_name) |
| 171 | |
| 172 | |
| 173 | # --------------------------------------------------------------------------- |
| 174 | # Synchronous runner (called from app.py @cli.command registration) |
| 175 | # --------------------------------------------------------------------------- |
| 176 | |
| 177 | |
| 178 | def run_checkout(*, branch: str, create: bool, force: bool) -> None: |
| 179 | """Synchronous entry point wired to the CLI by ``app.py``. |
| 180 | |
| 181 | Intentionally separated from the Typer decorator so that ``checkout`` |
| 182 | can be registered as a plain ``@cli.command()`` (a Click *Command*, not |
| 183 | a *Group*). Click Groups always invoke sub-contexts with |
| 184 | ``allow_interspersed_args=False``, which prevents options like |
| 185 | ``--force`` from being parsed when they appear *after* a positional |
| 186 | argument. Registering as a plain command avoids the issue entirely. |
| 187 | """ |
| 188 | root = require_repo() |
| 189 | |
| 190 | async def _run() -> None: |
| 191 | async with open_session() as session: |
| 192 | await _checkout_async( |
| 193 | branch_name=branch, |
| 194 | create=create, |
| 195 | force=force, |
| 196 | root=root, |
| 197 | session=session, |
| 198 | ) |
| 199 | |
| 200 | try: |
| 201 | asyncio.run(_run()) |
| 202 | except typer.Exit: |
| 203 | raise |
| 204 | except Exception as exc: |
| 205 | typer.echo(f"❌ muse checkout failed: {exc}") |
| 206 | logger.error("❌ muse checkout error: %s", exc, exc_info=True) |
| 207 | raise typer.Exit(code=ExitCode.INTERNAL_ERROR) |