artifact_resolver.py
python
| 1 | """Artifact resolution for ``muse open`` / ``muse play``. |
| 2 | |
| 3 | Resolves a user-supplied path-or-commit-ID to a concrete ``pathlib.Path``: |
| 4 | |
| 5 | - If the argument is an existing filesystem path (absolute or relative to |
| 6 | ``muse-work/``), return it directly — no DB needed. |
| 7 | - If the argument looks like a commit-ID prefix (4–64 lowercase hex chars), |
| 8 | query the DB for matching commits, present an interactive selection menu |
| 9 | when the snapshot contains multiple files, and return the resolved |
| 10 | working-tree path. |
| 11 | |
| 12 | The public async entry point ``resolve_artifact_async`` accepts an injected |
| 13 | ``AsyncSession`` so it can be unit-tested without a live database. |
| 14 | The synchronous wrapper ``resolve_artifact`` is suitable for use inside |
| 15 | Typer command callbacks. |
| 16 | """ |
| 17 | from __future__ import annotations |
| 18 | |
| 19 | import asyncio |
| 20 | import logging |
| 21 | import pathlib |
| 22 | |
| 23 | import typer |
| 24 | from sqlalchemy.ext.asyncio import AsyncSession |
| 25 | |
| 26 | from maestro.muse_cli.db import find_commits_by_prefix, open_session |
| 27 | from maestro.muse_cli.errors import ExitCode |
| 28 | from maestro.muse_cli.models import MuseCliCommit, MuseCliSnapshot |
| 29 | |
| 30 | logger = logging.getLogger(__name__) |
| 31 | |
| 32 | _HEX_CHARS = frozenset("0123456789abcdef") |
| 33 | |
| 34 | |
| 35 | def _looks_like_commit_prefix(s: str) -> bool: |
| 36 | """Return True if *s* could be a commit-ID prefix. |
| 37 | |
| 38 | Accepts 4–64 lower-case hex characters. Intentionally conservative: |
| 39 | actual filesystem paths that happen to be hex strings are excluded |
| 40 | early by the existence-check callers perform before calling this. |
| 41 | """ |
| 42 | lower = s.lower() |
| 43 | return 4 <= len(lower) <= 64 and all(c in _HEX_CHARS for c in lower) |
| 44 | |
| 45 | |
| 46 | |
| 47 | async def resolve_artifact_async( |
| 48 | path_or_commit_id: str, |
| 49 | root: pathlib.Path, |
| 50 | session: AsyncSession, |
| 51 | ) -> pathlib.Path: |
| 52 | """Resolve *path_or_commit_id* to a concrete working-tree path. |
| 53 | |
| 54 | Resolution order: |
| 55 | 1. Existing absolute/relative path on the filesystem. |
| 56 | 2. Path relative to ``<root>/muse-work/``. |
| 57 | 3. Commit-ID prefix lookup → interactive file selection from snapshot. |
| 58 | |
| 59 | Calls ``typer.Exit(ExitCode.USER_ERROR)`` on any user-facing error so |
| 60 | Typer surfaces a clean message instead of a traceback. |
| 61 | |
| 62 | Parameters |
| 63 | ---------- |
| 64 | path_or_commit_id: |
| 65 | Either a filesystem path or a hex commit-ID prefix (≥ 4 chars). |
| 66 | root: |
| 67 | The Muse repository root (containing ``.muse/`` and ``muse-work/``). |
| 68 | session: |
| 69 | An open ``AsyncSession`` — injected by callers for testability. |
| 70 | """ |
| 71 | # ── 1. Direct filesystem path ────────────────────────────────────────── |
| 72 | candidate = pathlib.Path(path_or_commit_id) |
| 73 | if candidate.exists(): |
| 74 | return candidate.resolve() |
| 75 | |
| 76 | # ── 2. Relative to muse-work/ ───────────────────────────────────────── |
| 77 | workdir_candidate = root / "muse-work" / path_or_commit_id |
| 78 | if workdir_candidate.exists(): |
| 79 | return workdir_candidate.resolve() |
| 80 | |
| 81 | # ── 3. Commit-ID prefix ─────────────────────────────────────────────── |
| 82 | prefix = path_or_commit_id.lower() |
| 83 | if not _looks_like_commit_prefix(prefix): |
| 84 | typer.echo(f"❌ File not found: {path_or_commit_id}") |
| 85 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 86 | |
| 87 | commits = await find_commits_by_prefix(session, prefix) |
| 88 | if not commits: |
| 89 | typer.echo(f"❌ No commit found matching prefix '{prefix[:8]}'") |
| 90 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 91 | |
| 92 | if len(commits) > 1: |
| 93 | typer.echo( |
| 94 | f"❌ Ambiguous commit prefix '{prefix[:8]}' — matches {len(commits)} commits:" |
| 95 | ) |
| 96 | for c in commits: |
| 97 | typer.echo(f" {c.commit_id[:8]} {c.message[:60]}") |
| 98 | typer.echo("Use a longer prefix to disambiguate.") |
| 99 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 100 | |
| 101 | commit = commits[0] |
| 102 | snapshot: MuseCliSnapshot | None = await session.get(MuseCliSnapshot, commit.snapshot_id) |
| 103 | if snapshot is None or not snapshot.manifest: |
| 104 | typer.echo(f"❌ Snapshot for commit {commit.commit_id[:8]} is empty.") |
| 105 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 106 | |
| 107 | manifest: dict[str, str] = snapshot.manifest |
| 108 | paths = sorted(manifest.keys()) |
| 109 | |
| 110 | if len(paths) == 1: |
| 111 | chosen = paths[0] |
| 112 | else: |
| 113 | typer.echo(f"Commit {commit.commit_id[:8]} — {commit.message}") |
| 114 | typer.echo("Files in this snapshot:") |
| 115 | for i, p in enumerate(paths, 1): |
| 116 | typer.echo(f" [{i}] {p}") |
| 117 | raw = typer.prompt("Select file number", default="1") |
| 118 | try: |
| 119 | idx = int(raw) - 1 |
| 120 | if idx < 0 or idx >= len(paths): |
| 121 | raise ValueError("out of range") |
| 122 | except ValueError: |
| 123 | typer.echo("❌ Invalid selection.") |
| 124 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 125 | chosen = paths[idx] |
| 126 | |
| 127 | resolved = root / "muse-work" / chosen |
| 128 | if not resolved.exists(): |
| 129 | typer.echo( |
| 130 | f"❌ '{chosen}' from commit {commit.commit_id[:8]} is no longer in muse-work/.\n" |
| 131 | " The snapshot references files that have been removed from the working tree." |
| 132 | ) |
| 133 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 134 | |
| 135 | logger.info("✅ Resolved '%s' → %s", path_or_commit_id, resolved) |
| 136 | return resolved.resolve() |
| 137 | |
| 138 | |
| 139 | def resolve_artifact( |
| 140 | path_or_commit_id: str, |
| 141 | root: pathlib.Path, |
| 142 | ) -> pathlib.Path: |
| 143 | """Synchronous wrapper around ``resolve_artifact_async``. |
| 144 | |
| 145 | Opens its own DB session via ``open_session()`` which reads |
| 146 | ``DATABASE_URL`` from settings. Suitable for use in Typer command |
| 147 | callbacks that need a blocking call. |
| 148 | """ |
| 149 | |
| 150 | async def _run() -> pathlib.Path: |
| 151 | async with open_session() as session: |
| 152 | return await resolve_artifact_async( |
| 153 | path_or_commit_id, root=root, session=session |
| 154 | ) |
| 155 | |
| 156 | return asyncio.run(_run()) |