"""``muse mist`` — create, share, and manage content-addressed Muse Mists. A *Mist* is the Muse answer to GitHub gists: a single artifact (code, MIDI, prose, schema, ABI, or any binary blob) captured in the Muse object store, content-addressed by its SHA-256 digest, signed with an Ed25519 key, and version-controlled via a lightweight Muse repo with ``domain="mist"``. Unlike a gist, a Mist: - Has a globally unique, human-readable 12-character ID derived from content. - Carries author provenance: Ed25519 signature + optional agent_id/model_id. - Has full VCS lineage: branches, commits, proposals, diffs, releases. - Is forkable with proposal-back-to-upstream support. - Is embeddable via ``/embed`` with domain-appropriate rendering. - Is MCP-accessible as ``muse:///handle/mists/ID``. Subcommands ----------- create Create a new Mist from a local file. list List Mists for the authenticated user or a given handle. read Read a Mist's content and metadata. fork Fork a Mist into the caller's namespace. update Update a Mist's title, description, visibility, tags, or content. forks List direct forks of a Mist. raw Print or save the raw artifact bytes of a Mist. push Push a local Mist repo to MuseHub. embed Generate embed code for a Mist. delete Delete a Mist (owner only). All subcommands accept ``--json`` for machine-readable output. ``create`` additionally accepts ``--sign`` to attach the caller's Ed25519 signature, and ``--push`` to submit to MuseHub immediately after creation. Exit codes ---------- 0 Success. 1 User error — invalid arguments or bad input. 2 Not inside a Muse repository (for ``push`` subcommand). 3 File not found or unreadable. 4 Mist not found on MuseHub. 5 Permission denied (for ``delete``). JSON output example (``create --json``):: { "mist_id": "aB3xKq9dPwNm", "url": "https://musehub.ai/gabriel/mists/aB3xKq9dPwNm", "artifact_type": "code", "language": "python", "filename": "validate_assignee.py", "size_bytes": 892, "signed": true, "agent_id": "", "model_id": "" } """ import argparse import json import os import sys import urllib.parse from collections.abc import Mapping from muse.core.envelope import JsonValue from muse.core.errors import ExitCode from muse.core.identity import IdentityEntry, load_identity from muse.core.validation import sanitize_display from muse.plugins.mist.plugin import ( MIST_VISIBILITIES, compute_mist_id, detect_artifact_type, extract_mist_symbol_anchors, validate_mist_filename, ) # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- _MAX_MIST_BYTES = 10 * 1024 * 1024 # 10 MiB hard limit _MAX_TAG_LENGTH = 64 _MAX_TAGS = 10 _ALLOWED_API_SCHEMES = frozenset({"http", "https"}) # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _get_hub_url() -> tuple[str, IdentityEntry] | None: """Return (hub_url, identity) for the current context, or None. Tries the hub URL from ``.muse/config.toml``, then falls back to the ``local`` remote. Returns ``None`` when neither is available — callers that require MuseHub print their own error and exit. Returns: A ``(hub_url, identity)`` tuple, or ``None`` if no hub is available. """ try: from muse.cli.config import get_hub_url, get_remote, list_remotes from muse.core.repo import find_repo_root root = find_repo_root() hub_url: str | None = None if root is not None: hub_url = get_hub_url(root) if hub_url is None: remote_url = get_remote("local", root) if remote_url: hub_url = remote_url.rstrip("/") else: remotes = list_remotes(root) if remotes: hub_url = remotes[0]["url"].rstrip("/") if hub_url: identity = load_identity(hub_url) if identity: return hub_url, identity return None except Exception: return None type _JsonObject = dict[str, JsonValue] def _hub_api( hub_url: str, identity: IdentityEntry, method: str, path: str, body: Mapping[str, JsonValue] | None = None, hub_override: str | None = None, timeout: float = 15.0, ) -> _JsonObject: """Make an authenticated JSON request to the MuseHub API. Uses :class:`~muse.core.transport.HttpTransport` (httpx + mkcert) so that self-signed localhost certificates are handled correctly. Args: hub_url: Repository-level or server-root hub URL. identity: Loaded identity entry for signing. method: HTTP method (``GET``, ``POST``, ``PATCH``, ``DELETE``). path: API path (e.g. ``/api/mists/{id}``). body: Optional JSON body dict. hub_override: Override the server root (from ``--hub`` flag). timeout: Ignored — transport uses its own timeout configuration. Returns: Parsed JSON response as a dict. Raises: SystemExit: On scheme error, auth error, network error, or non-2xx response. """ from muse.cli.config import get_signing_identity from muse.core.transport import HttpTransport, TransportError root_url = hub_override or hub_url parsed = urllib.parse.urlparse(root_url) scheme = parsed.scheme.lower() if scheme not in _ALLOWED_API_SCHEMES: print( f"❌ Hub URL scheme {sanitize_display(scheme)!r} is not allowed. " "Use http or https.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) server_root = f"{parsed.scheme}://{parsed.netloc}" url = f"{server_root}{path}" signing = get_signing_identity(remote_url=server_root) try: return HttpTransport().hub_json(method, url, signing, body=dict(body) if body is not None else None) except TransportError as exc: status = exc.status_code detail = sanitize_display(str(exc)) if status == 401: print("❌ Not authenticated. Run: muse auth register", file=sys.stderr) elif status == 403: print(f"❌ Permission denied: {detail or path}", file=sys.stderr) raise SystemExit(ExitCode.REMOTE_ERROR) elif status == 404: print(f"❌ Not found: {detail or path}", file=sys.stderr) raise SystemExit(ExitCode.NOT_FOUND) elif status == 413: print("❌ Content too large (limit: 10 MiB).", file=sys.stderr) elif status == 422: print(f"❌ Validation error: {detail}", file=sys.stderr) else: print(f"❌ Hub returned HTTP {status}: {detail}", file=sys.stderr) raise SystemExit(ExitCode.REMOTE_ERROR) def _require_hub(hub_override: str | None = None) -> tuple[str, IdentityEntry]: """Return (hub_url, identity) or exit with a clear error. Accepts an explicit ``--hub URL`` override; otherwise resolves from the repo config and falls back to the ``local`` remote. Args: hub_override: Optional ``--hub`` flag value. Returns: A ``(hub_url, identity)`` tuple. Raises: SystemExit: If no hub URL is available or the user is not authenticated. """ if hub_override: identity = load_identity(hub_override) if not identity: print("❌ Not authenticated. Run: muse auth register", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) return hub_override.rstrip("/"), identity ctx = _get_hub_url() if ctx is None: print( "❌ No MuseHub configured. Run: muse hub connect ", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) return ctx def _validate_tag(tag: str) -> None: """Validate a single mist tag string. Tags must be non-empty, ≤ 64 characters, contain no control characters, no HTML-special characters, and no null bytes. Args: tag: The tag string to validate. Raises: ValueError: With a description of the violation. """ if not tag or not tag.strip(): raise ValueError("Tags must be non-empty strings.") if len(tag) > _MAX_TAG_LENGTH: raise ValueError(f"Tag exceeds {_MAX_TAG_LENGTH}-character limit: {tag!r}") if "\x00" in tag: raise ValueError(f"Tag must not contain null bytes: {tag!r}") for ch in tag: cp = ord(ch) if 0x01 <= cp <= 0x1F or cp == 0x7F: raise ValueError(f"Tag must not contain control characters: {tag!r}") for bad in ("<", ">", '"', "'", "&"): if bad in tag: raise ValueError(f"Tag must not contain HTML special character {bad!r}: {tag!r}") # --------------------------------------------------------------------------- # Subcommand handlers # --------------------------------------------------------------------------- def run_create(args: argparse.Namespace) -> None: """Create a new Mist from a local file. Reads the file at ``FILE``, validates the filename, computes a content-addressed ``mist_id`` (12-character base-58 SHA-256 prefix), detects the artifact type and language, and extracts symbol anchors for code artifacts. With ``--push``, the mist is submitted to MuseHub via ``POST /api/mists``. Without ``--push``, only local metadata is computed and printed — useful for preview and scripting. Signing (``--sign``) attaches the caller's Ed25519 signature from ``~/.muse/identity.toml``. AI agents set ``--agent-id`` and ``--model-id`` for provenance tracking. JSON output (stdout) when ``--json`` ------------------------------------ :: { "mist_id": "aB3xKq9dPwNm", "url": "https://musehub.ai/gabriel/mists/aB3xKq9dPwNm", "artifact_type": "code", "language": "python", "filename": "validate_assignee.py", "size_bytes": 892, "signed": true, "agent_id": "cccode-v3", "model_id": "claude-sonnet-4-6", "symbol_anchors": ["validate_assignee.py::_validate_assignee"] } Exit codes ---------- 0 Success. 1 User error (invalid filename, tag, or visibility value). 3 File not found or unreadable. 5 MuseHub API error (when --push). Args: args: Parsed argument namespace from the ``create`` subparser. """ file_path: str = args.file json_output: bool = args.json_output do_push: bool = args.push do_sign: bool = args.sign title: str = args.title or "" description: str = args.description or "" visibility: str = args.visibility or "public" tag_strings: list[str] = args.tags or [] agent_id: str = args.agent_id or "" model_id: str = args.model_id or "" hub_override: str | None = getattr(args, "hub", None) # Validate visibility if visibility not in MIST_VISIBILITIES: print( f"❌ Invalid visibility {visibility!r}. Choose: public, secret", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # Validate tags if len(tag_strings) > _MAX_TAGS: print(f"❌ Too many tags (max {_MAX_TAGS}): {len(tag_strings)} given.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) for tag in tag_strings: try: _validate_tag(tag) except ValueError as exc: print(f"❌ {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # Read file try: with open(file_path, "rb") as fh: content = fh.read() except FileNotFoundError: print(f"❌ File not found: {sanitize_display(file_path)}", file=sys.stderr) raise SystemExit(ExitCode.NOT_FOUND) except PermissionError: print(f"❌ Permission denied: {sanitize_display(file_path)}", file=sys.stderr) raise SystemExit(ExitCode.REMOTE_ERROR) except OSError as exc: print(f"❌ Cannot read file: {sanitize_display(str(exc))}", file=sys.stderr) raise SystemExit(ExitCode.NOT_FOUND) if len(content) > _MAX_MIST_BYTES: print( f"❌ File exceeds 10 MiB limit: {len(content):,} bytes.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) filename = os.path.basename(file_path) try: validate_mist_filename(filename) except ValueError as exc: print(f"❌ {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # Compute mist properties mist_id = compute_mist_id(content) type_info = detect_artifact_type(filename, content) artifact_type = type_info["artifact_type"] language = type_info["language"] size_bytes = len(content) symbol_anchors = extract_mist_symbol_anchors(filename, content) # Sign if requested gpg_signature: str | None = None signed = False if do_sign: try: from muse.cli.config import get_signing_identity from muse.core.keypair import sign_bytes as _sign_bytes _signing = get_signing_identity() if _signing: gpg_signature = _sign_bytes(_signing.private_key, content) signed = True except Exception as exc: print( f"⚠️ Could not sign mist: {sanitize_display(str(exc))}", file=sys.stderr, ) # Build content string (base64 for binary, utf-8 for text) content_str: str try: content_str = content.decode("utf-8") except UnicodeDecodeError: import base64 content_str = base64.b64encode(content).decode("ascii") url = "" if do_push: hub_url, identity = _require_hub(hub_override) # Derive server root from hub_url (strip repo path if present) parsed = urllib.parse.urlparse(hub_url) server_root = f"{parsed.scheme}://{parsed.netloc}" handle = identity.get("handle", "") payload = { "filename": filename, "content": content_str, "artifact_type": artifact_type, "language": language, "title": title, "description": description, "visibility": visibility, "tags": tag_strings, "agent_id": agent_id, "model_id": model_id, } if gpg_signature: payload["gpg_signature"] = gpg_signature data = _hub_api(server_root, identity, "POST", "/api/mists", body=payload) mist_id = str(data.get("mist_id", mist_id)) handle = str(data.get("owner", handle)) url = f"{server_root}/{handle}/mists/{mist_id}" result = { "mist_id": mist_id, "url": url, "artifact_type": artifact_type, "language": language, "filename": filename, "size_bytes": size_bytes, "signed": signed, "agent_id": agent_id, "model_id": model_id, "symbol_anchors": symbol_anchors, } if json_output: print(json.dumps(result)) return type_badge = f"[{artifact_type}]" if artifact_type != "unknown" else "[unknown type]" lang_badge = f" [{language}]" if language else "" sign_badge = " [signed ✓]" if signed else "" print(f"✅ Mist created") print(f" ID: {mist_id}") print(f" File: {sanitize_display(filename)}") print(f" Type: {type_badge}{lang_badge}{sign_badge}") print(f" Size: {size_bytes:,} bytes") if symbol_anchors: print(f" Symbols: {len(symbol_anchors)} ({', '.join(symbol_anchors[:3])}{'…' if len(symbol_anchors) > 3 else ''})") if url: print(f" URL: {url}") else: print(" (Use --push to publish to MuseHub)") def run_list(args: argparse.Namespace) -> None: """List Mists for a MuseHub handle. Queries ``GET /api/{handle}/mists`` on MuseHub. When ``--handle`` is omitted, uses the authenticated user's handle from ``~/.muse/identity.toml``. Pagination is cursor-based: each response includes a ``next_cursor`` field. Pass it with ``--cursor`` to retrieve the next page. JSON output (stdout) when ``--json`` ------------------------------------ :: { "total": 47, "next_cursor": "cursor_string_or_null", "mists": [ { "mist_id": "aB3xKq9dPwNm", "owner": "gabriel", "artifact_type": "code", "language": "python", "filename": "validate_assignee.py", "title": "...", "size_bytes": 892, "signed": true, "fork_count": 3, "view_count": 842, "visibility": "public", "tags": [], "version": 3, "created_at": "2026-04-14T00:00:00Z", "updated_at": "2026-04-14T00:00:00Z" } ] } Args: args: Parsed argument namespace from the ``list`` subparser. """ handle: str | None = args.handle json_output: bool = args.json_output limit: int = max(1, min(args.limit, 100)) cursor: str | None = args.cursor artifact_type_filter: str | None = args.type hub_override: str | None = getattr(args, "hub", None) hub_url, identity = _require_hub(hub_override) parsed = urllib.parse.urlparse(hub_url) server_root = f"{parsed.scheme}://{parsed.netloc}" if not handle: handle = identity.get("handle", "") if not handle: print("❌ No handle provided and no identity configured.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) params: dict[str, str] = {"limit": str(limit)} if cursor: params["cursor"] = cursor if artifact_type_filter: params["artifact_type"] = artifact_type_filter query_string = "&".join(f"{k}={urllib.parse.quote(v)}" for k, v in params.items()) api_path = f"/api/{urllib.parse.quote(handle)}/mists?{query_string}" data = _hub_api(server_root, identity, "GET", api_path) if json_output: print(json.dumps(data)) return mists: list[dict] = data.get("mists", []) total: int = data.get("total", len(mists)) next_cursor: str | None = data.get("next_cursor") if not mists: print(f" {sanitize_display(handle)} has no mists.") return print(f" {sanitize_display(handle)} / mists ({total} total)") print() for m in mists: mid = sanitize_display(str(m.get("mist_id", ""))) atype = m.get("artifact_type", "unknown") lang = m.get("language", "") fname = sanitize_display(str(m.get("filename", ""))) ttl = sanitize_display(str(m.get("title", ""))) forks = m.get("fork_count", 0) views = m.get("view_count", 0) vis = m.get("visibility", "public") signed = m.get("signed", False) badges = f"[{atype}]" if lang: badges += f" [{lang}]" if signed: badges += " [signed]" if vis == "secret": badges += " [secret]" label = ttl or fname or mid print(f" {mid} {badges}") print(f" {label}") print(f" {views} views · {forks} forks") print() if next_cursor: print(f" (More results — use --cursor {next_cursor!r} for next page)") def run_read(args: argparse.Namespace) -> None: """Read a Mist's content and metadata from MuseHub. Resolves the mist by ``MIST_ID`` (12-character base-58 ID or ``owner/ID`` form). Increments the view count on the server. JSON output (stdout) when ``--json`` ------------------------------------ :: { "mist_id": "aB3xKq9dPwNm", "url": "https://musehub.ai/gabriel/mists/aB3xKq9dPwNm", "owner": "gabriel", "artifact_type": "code", "language": "python", "filename": "validate_assignee.py", "title": "...", "description": "...", "content": "def _validate_assignee...", "size_bytes": 892, "signed": true, "agent_id": "", "model_id": "", "fork_count": 3, "view_count": 843, "visibility": "public", "tags": [], "version": 3, "symbol_anchors": ["validate_assignee.py::_validate_assignee"], "created_at": "2026-04-14T00:00:00Z", "updated_at": "2026-04-14T00:00:00Z" } Args: args: Parsed argument namespace from the ``read`` subparser. """ mist_id: str = args.mist_id.strip() json_output: bool = args.json_output hub_override: str | None = getattr(args, "hub", None) hub_url, identity = _require_hub(hub_override) parsed = urllib.parse.urlparse(hub_url) server_root = f"{parsed.scheme}://{parsed.netloc}" # Support owner/ID form if "/" in mist_id: parts = mist_id.split("/", 1) owner_part = urllib.parse.quote(parts[0].strip()) id_part = urllib.parse.quote(parts[1].strip()) api_path = f"/api/{owner_part}/mists/{id_part}" else: api_path = f"/api/mists/{urllib.parse.quote(mist_id)}" data = _hub_api(server_root, identity, "GET", api_path) if json_output: print(json.dumps(data)) return mid = sanitize_display(str(data.get("mist_id", mist_id))) owner = sanitize_display(str(data.get("owner", ""))) atype = sanitize_display(str(data.get("artifact_type", "unknown"))) lang = sanitize_display(str(data.get("language", ""))) fname = sanitize_display(str(data.get("filename", ""))) ttl = sanitize_display(str(data.get("title", ""))) signed = data.get("signed", False) agent_id = sanitize_display(str(data.get("agent_id", ""))) model_id = sanitize_display(str(data.get("model_id", ""))) version = data.get("version", 1) forks = data.get("fork_count", 0) views = data.get("view_count", 0) content = data.get("content", "") anchors: list[str] = data.get("symbol_anchors", []) print(f" {owner} / mists / {mid}") if ttl: print(f" \"{sanitize_display(ttl)}\"") print(f" [{atype}]{' [' + lang + ']' if lang else ''}{' [signed ✓]' if signed else ''}") if agent_id: print(f" Agent: {agent_id} Model: {model_id}") print(f" v{version} · {views} views · {forks} forks") if anchors: print(f" Symbols: {', '.join(anchors[:5])}{'…' if len(anchors) > 5 else ''}") print() # Print first 40 lines of content for human-readable preview lines = content.splitlines() preview_lines = lines[:40] for line in preview_lines: print(f" {sanitize_display(line)}") if len(lines) > 40: print(f" … ({len(lines) - 40} more lines — use --json for full content)") def run_fork(args: argparse.Namespace) -> None: """Fork a Mist into the caller's namespace. Creates a new Mist in the caller's namespace rooted at the same commit as the original. Sets ``fork_parent_id`` on the new mist to the original's ``mist_id``. Increments ``fork_count`` on the original. JSON output (stdout) when ``--json`` ------------------------------------ :: { "mist_id": "Kx2mPq7bRnYt", "url": "https://musehub.ai/you/mists/Kx2mPq7bRnYt", "fork_parent_id": "aB3xKq9dPwNm", "owner": "you", "artifact_type": "code", "language": "python" } Args: args: Parsed argument namespace from the ``fork`` subparser. """ mist_id: str = args.mist_id.strip() json_output: bool = args.json_output hub_override: str | None = getattr(args, "hub", None) hub_url, identity = _require_hub(hub_override) parsed = urllib.parse.urlparse(hub_url) server_root = f"{parsed.scheme}://{parsed.netloc}" if "/" in mist_id: parts = mist_id.split("/", 1) id_part = urllib.parse.quote(parts[1].strip()) else: id_part = urllib.parse.quote(mist_id) api_path = f"/api/mists/{id_part}/fork" data = _hub_api(server_root, identity, "POST", api_path) if json_output: print(json.dumps(data)) return new_id = sanitize_display(str(data.get("mist_id", ""))) owner = sanitize_display(str(data.get("owner", identity.get("handle", "")))) url = data.get("url", f"{server_root}/{owner}/mists/{new_id}") print(f"✅ Mist forked") print(f" New ID: {new_id}") print(f" Owner: {owner}") print(f" URL: {sanitize_display(url)}") print(f" Parent: {sanitize_display(mist_id)}") def run_push(args: argparse.Namespace) -> None: """Push a local Mist repo to MuseHub. Must be run from inside a Muse repository with ``domain="mist"``. Wraps the standard ``muse push`` infrastructure — the remote name defaults to ``local`` but can be overridden with ``--remote``. This is the multi-step workflow alternative to ``muse mist create --push``: 1. ``muse init --domain mist`` 2. Add your artifact file and ``muse commit`` 3. ``muse mist push [--remote local]`` Exit codes ---------- 0 Success. 2 Not inside a Muse repository or domain is not "mist". Args: args: Parsed argument namespace from the ``push`` subparser. """ remote: str = args.remote or "local" branch: str = args.branch or "main" json_output: bool = args.json_output try: from muse.core.repo import find_repo_root except ImportError: print("❌ Cannot import muse repo utilities.", file=sys.stderr) raise SystemExit(ExitCode.INTERNAL_ERROR) root = find_repo_root() if root is None: print("❌ Not inside a Muse repository.", file=sys.stderr) raise SystemExit(ExitCode.REPO_NOT_FOUND) # Verify domain is "mist" from muse.plugins.registry import read_domain domain = read_domain(root) if domain != "mist": print( f"❌ This repo has domain={domain!r}, not 'mist'. " "Run from inside a mist repo (muse init --domain mist).", file=sys.stderr, ) raise SystemExit(ExitCode.REPO_NOT_FOUND) # Delegate to the push command's internals from muse.cli.commands.push import run as push_run import types push_args = types.SimpleNamespace( remote=remote, branch=branch, force=False, json_output=json_output, ) push_run(push_args) def run_embed(args: argparse.Namespace) -> None: """Generate embed code for a Mist. Returns HTML iframe, JavaScript snippet, and Markdown badge code for embedding a Mist in external pages, documentation, or dashboards. JSON output (stdout) when ``--json`` ------------------------------------ :: { "mist_id": "aB3xKq9dPwNm", "owner": "gabriel", "iframe": "", "js": "", "badge": "[![Mist](...)](/gabriel/mists/aB3xKq9dPwNm)" } Args: args: Parsed argument namespace from the ``embed`` subparser. """ mist_id: str = args.mist_id.strip() json_output: bool = args.json_output width: int = max(200, min(args.width, 1920)) height: int = max(100, min(args.height, 1080)) hub_override: str | None = getattr(args, "hub", None) hub_url, identity = _require_hub(hub_override) parsed = urllib.parse.urlparse(hub_url) server_root = f"{parsed.scheme}://{parsed.netloc}" if "/" in mist_id: parts = mist_id.split("/", 1) owner_part = urllib.parse.quote(parts[0].strip()) id_part = urllib.parse.quote(parts[1].strip()) else: owner_part = urllib.parse.quote(identity.get("handle", "")) id_part = urllib.parse.quote(mist_id) api_path = f"/api/{owner_part}/mists/{id_part}/embed" data = _hub_api(server_root, identity, "GET", api_path) owner = data.get("owner", owner_part) clean_id = sanitize_display(str(data.get("mist_id", mist_id))) embed_url = f"{server_root}/{sanitize_display(owner)}/mists/{clean_id}/embed" iframe = ( data.get("iframe") or f'' ) js = ( data.get("js") or f'' ) page_url = f"{server_root}/{sanitize_display(owner)}/mists/{clean_id}" badge = ( data.get("badge") or f'[![Mist {clean_id}]({server_root}/static/badge.svg)]({page_url})' ) result = { "mist_id": clean_id, "owner": sanitize_display(str(owner)), "iframe": iframe, "js": js, "badge": badge, } if json_output: print(json.dumps(result)) return print(f" Embed code for mist {clean_id}") print() print(" iframe:") print(f" {iframe}") print() print(" JS snippet:") print(f" {js}") print() print(" Markdown badge:") print(f" {badge}") def run_delete(args: argparse.Namespace) -> None: """Delete a Mist from MuseHub (owner only). Sends ``DELETE /api/mists/{id}`` to MuseHub. Requires ownership — returns HTTP 403 for non-owners. The ``--yes`` flag skips the interactive confirmation prompt. This operation is irreversible. The underlying Muse repo is also deleted. Exit codes ---------- 0 Success. 4 Mist not found. 5 Permission denied (not the owner). Args: args: Parsed argument namespace from the ``delete`` subparser. """ mist_id: str = args.mist_id.strip() yes: bool = args.yes json_output: bool = args.json_output hub_override: str | None = getattr(args, "hub", None) hub_url, identity = _require_hub(hub_override) parsed = urllib.parse.urlparse(hub_url) server_root = f"{parsed.scheme}://{parsed.netloc}" if "/" in mist_id: parts = mist_id.split("/", 1) id_part = urllib.parse.quote(parts[1].strip()) else: id_part = urllib.parse.quote(mist_id) if not yes: try: answer = input( f"Delete mist {sanitize_display(mist_id)}? This cannot be undone. [y/N] " ).strip().lower() except (EOFError, KeyboardInterrupt): print("\nAborted.", file=sys.stderr) raise SystemExit(0) if answer not in ("y", "yes"): print("Aborted.", file=sys.stderr) raise SystemExit(0) api_path = f"/api/mists/{id_part}" _hub_api(server_root, identity, "DELETE", api_path) result = {"mist_id": mist_id, "deleted": True} if json_output: print(json.dumps(result)) else: print(f"✅ Mist {sanitize_display(mist_id)} deleted.") def run_update(args: argparse.Namespace) -> None: """Update a Mist's metadata or replace its artifact content. Sends ``PATCH /api/mists/{mist_id}`` with only the fields that were explicitly supplied. Omitted flags are not sent — the server performs a partial update so unspecified fields remain unchanged. When ``--content FILE`` is supplied, the file is read as UTF-8 and its text replaces the current artifact. The server increments the mist's version counter on every content change. JSON output (stdout) when ``--json`` ------------------------------------ :: { "mist_id": "aB3xKq9dPwNm", "version": 2, "title": "Updated title", "visibility": "public", "updated_at": "2026-04-15T13:00:00+00:00" } Exit codes ---------- 0 Success. 1 User error — no fields supplied, or invalid visibility value. 4 Mist not found or caller is not the owner (HTTP 404). 5 Remote error — unexpected HTTP status. Args: args: Parsed argument namespace from the ``update`` subparser. Relevant attributes: ``mist_id``, ``title``, ``description``, ``visibility``, ``tags``, ``content``, ``hub``, ``json_output``. """ import pathlib mist_id: str = args.mist_id.strip() json_output: bool = args.json_output hub_override: str | None = getattr(args, "hub", None) payload = {} if args.title is not None: payload["title"] = args.title if args.description is not None: payload["description"] = args.description if args.visibility is not None: if args.visibility not in MIST_VISIBILITIES: print( f"❌ Invalid visibility {args.visibility!r}. Choose: public, secret", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) payload["visibility"] = args.visibility if args.tags is not None: payload["tags"] = [t.strip() for t in args.tags.split(",") if t.strip()] if args.content is not None: try: content_path = pathlib.Path(args.content) payload["content"] = content_path.read_text(encoding="utf-8") payload["filename"] = content_path.name except OSError as exc: print(f"❌ Cannot read content file: {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if not payload: print( "❌ Nothing to update — provide at least one of: " "--title, --description, --visibility, --tags, --content", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) hub_url, identity = _require_hub(hub_override) parsed = urllib.parse.urlparse(hub_url) server_root = f"{parsed.scheme}://{parsed.netloc}" if "/" in mist_id: id_part = urllib.parse.quote(mist_id.split("/", 1)[1].strip()) else: id_part = urllib.parse.quote(mist_id) data = _hub_api(server_root, identity, "PATCH", f"/api/mists/{id_part}", body=payload) if json_output: print(json.dumps(data)) return clean_id = sanitize_display(str(data.get("mist_id", mist_id))) version = data.get("version", "?") print(f"✅ Mist {clean_id} updated (v{version})") if "title" in data: print(f" Title: {sanitize_display(str(data['title']))}") if "visibility" in data: print(f" Visibility: {sanitize_display(str(data['visibility']))}") def run_forks(args: argparse.Namespace) -> None: """List the direct (one-level) forks of a Mist. Calls ``GET /api/mists/{mist_id}/forks`` and renders each fork as a compact summary row. With ``--json``, prints the raw API response. JSON output (stdout) when ``--json`` ------------------------------------ :: [ { "mist_id": "Kx2mPq7bRnYt", "owner": "alice", "filename": "validate.py", "fork_depth": 1, "created_at": "2026-04-15T12:00:00+00:00" } ] Exit codes ---------- 0 Success (empty list is also a success). 4 Mist not found (HTTP 404). 5 Remote error — unexpected HTTP status. Args: args: Parsed argument namespace from the ``forks`` subparser. Relevant attributes: ``mist_id``, ``limit``, ``hub``, ``json_output``. """ mist_id: str = args.mist_id.strip() limit: int = max(1, min(args.limit, 100)) json_output: bool = args.json_output hub_override: str | None = getattr(args, "hub", None) hub_url, identity = _require_hub(hub_override) parsed = urllib.parse.urlparse(hub_url) server_root = f"{parsed.scheme}://{parsed.netloc}" if "/" in mist_id: id_part = urllib.parse.quote(mist_id.split("/", 1)[1].strip()) else: id_part = urllib.parse.quote(mist_id) api_path = f"/api/mists/{id_part}/forks?limit={limit}" data = _hub_api(server_root, identity, "GET", api_path) if json_output: print(json.dumps(data)) return forks: list[dict] = data if isinstance(data, list) else data.get("forks", []) if not forks: print(f" No forks for mist {sanitize_display(mist_id)}.") return print(f" Forks of {sanitize_display(mist_id)} ({len(forks)} shown):") for fork in forks: fid = sanitize_display(str(fork.get("mist_id", ""))) owner = sanitize_display(str(fork.get("owner", ""))) filename = sanitize_display(str(fork.get("filename", ""))) depth = fork.get("fork_depth", "?") print(f" {fid} {owner}/{filename} depth={depth}") def run_raw(args: argparse.Namespace) -> None: """Print or save the raw artifact bytes of a Mist. Calls ``GET /api/mists/{mist_id}/raw`` and streams the response bytes to stdout, or writes them to ``--output FILE``. Useful for piping directly into tools:: muse mist raw aB3xKq9dPwNm > validate_handle.py muse mist raw aB3xKq9dPwNm | python3 -c "import sys; exec(sys.stdin.read())" Exit codes ---------- 0 Success. 4 Mist not found (HTTP 404). 5 Permission denied — secret mist and not authenticated (HTTP 403), or other remote error. Args: args: Parsed argument namespace from the ``raw`` subparser. Relevant attributes: ``mist_id``, ``output``, ``hub``. """ import pathlib mist_id: str = args.mist_id.strip() output: str | None = getattr(args, "output", None) hub_override: str | None = getattr(args, "hub", None) hub_url, identity = _require_hub(hub_override) parsed = urllib.parse.urlparse(hub_url) server_root = f"{parsed.scheme}://{parsed.netloc}" if "/" in mist_id: id_part = urllib.parse.quote(mist_id.split("/", 1)[1].strip()) else: id_part = urllib.parse.quote(mist_id) # Build a raw-bytes request — Accept: */* so the server sends the artifact MIME type. from muse.cli.config import get_signing_identity from muse.core.transport import HttpTransport, TransportError url = f"{server_root}/api/mists/{id_part}/raw" signing = get_signing_identity(remote_url=server_root) try: raw_bytes: bytes = HttpTransport().hub_bytes(url, signing) except TransportError as exc: if exc.status_code == 404: print(f"❌ Mist not found: {sanitize_display(mist_id)}", file=sys.stderr) raise SystemExit(ExitCode.NOT_FOUND) if exc.status_code == 403: print("❌ Permission denied — secret mist or not authenticated.", file=sys.stderr) raise SystemExit(ExitCode.REMOTE_ERROR) print(f"❌ HTTP {exc.status_code} from hub.", file=sys.stderr) raise SystemExit(ExitCode.REMOTE_ERROR) if output: try: pathlib.Path(output).write_bytes(raw_bytes) print(f"✅ Saved {len(raw_bytes)} bytes to {sanitize_display(output)}") except OSError as exc: print(f"❌ Cannot write output file: {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) else: sys.stdout.buffer.write(raw_bytes) # --------------------------------------------------------------------------- # Subcommand registration # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``muse mist`` subcommand tree and all its flags. Subcommands ----------- create Create a new Mist from a local file. list List Mists for the authenticated user or a given handle. read Read a Mist's content and metadata. fork Fork a Mist into the caller's namespace. update Update a Mist's title, description, visibility, tags, or content. forks List direct forks of a Mist. raw Print or save the raw artifact bytes of a Mist. push Push a local Mist repo to MuseHub. embed Generate embed code for a Mist. delete Delete a Mist (owner only). All subcommands accept ``--json`` for machine-readable output. ``create`` additionally accepts ``--sign`` to attach the caller's Ed25519 signature, and ``--push`` to submit to MuseHub immediately after creation. Exit codes ---------- 0 Success. 1 User error — invalid arguments or bad input. 2 Not inside a Muse repository (for ``push``). 3 File not found or unreadable. 4 Mist not found on MuseHub. 5 Permission denied (for ``delete``). Args: subparsers: The top-level argument parser's subparsers action. """ parser = subparsers.add_parser( "mist", help="Create, share, and manage content-addressed Muse Mists.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) subs = parser.add_subparsers(dest="mist_subcommand", metavar="SUBCOMMAND") subs.required = True # ── create ──────────────────────────────────────────────────────────────── create_p = subs.add_parser( "create", help="Create a new Mist from a local file.", description=( "Read FILE and create a content-addressed Mist.\n\n" "The mist_id is the first 12 characters of the base-58 encoding of\n" "the file's SHA-256 digest — same bytes always yield the same ID.\n\n" "Without --push, only local metadata is computed (no network required).\n" "With --push, the Mist is submitted to MuseHub via POST /api/mists.\n\n" "Agent quickstart:\n" " muse mist create script.py --sign --push --json\n" " muse mist create track.mid --title 'My motif' --push --json" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) create_p.add_argument("file", metavar="FILE", help="Path to the artifact file.") create_p.add_argument( "--title", "-t", metavar="TEXT", default="", help="Optional human-readable title for the Mist.", ) create_p.add_argument( "--description", "-d", metavar="TEXT", default="", help="Optional Markdown description.", ) create_p.add_argument( "--visibility", metavar="public|secret", default="public", help="Visibility: 'public' (default) or 'secret' (direct-URL only).", ) create_p.add_argument( "--tag", dest="tags", action="append", default=[], metavar="TAG", help="Add a tag (repeatable, max 10).", ) create_p.add_argument( "--sign", action="store_true", default=False, help="Sign the Mist with the caller's Ed25519 key from identity.toml.", ) create_p.add_argument( "--push", action="store_true", default=False, help="Publish the Mist to MuseHub immediately after creation.", ) create_p.add_argument( "--agent-id", dest="agent_id", metavar="ID", default="", help="MSign agent identifier (set automatically in agent contexts).", ) create_p.add_argument( "--model-id", dest="model_id", metavar="ID", default="", help="Model identifier for AI provenance (e.g. claude-sonnet-4-6).", ) create_p.add_argument( "--hub", metavar="URL", default=None, help="Override the MuseHub URL (default: from .muse/config.toml).", ) create_p.add_argument( "--json", "-j", action="store_true", dest="json_output", default=False, help="Emit a JSON object to stdout on success.", ) create_p.set_defaults(func=run_create) # ── list ────────────────────────────────────────────────────────────────── list_p = subs.add_parser( "list", help="List Mists for the authenticated user or a given handle.", description=( "List Mists on MuseHub. Defaults to the authenticated user's Mists.\n\n" "Agent quickstart:\n" " muse mist list --json\n" " muse mist list --handle gabriel --type code --json" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) list_p.add_argument( "--handle", "-u", metavar="HANDLE", default=None, help="MuseHub handle to list Mists for (default: authenticated user).", ) list_p.add_argument( "--type", metavar="TYPE", default=None, help="Filter by artifact_type (code, midi, prose, schema, abi, unknown).", ) list_p.add_argument( "--limit", "-n", type=int, default=20, metavar="N", help="Maximum number of Mists to return per page (default: 20, max: 100).", ) list_p.add_argument( "--cursor", metavar="CURSOR", default=None, help="Pagination cursor from a previous list response.", ) list_p.add_argument( "--hub", metavar="URL", default=None, help="Override the MuseHub URL.", ) list_p.add_argument( "--json", "-j", action="store_true", dest="json_output", default=False, help="Emit a JSON object to stdout.", ) list_p.set_defaults(func=run_list) # ── read ────────────────────────────────────────────────────────────────── read_p = subs.add_parser( "read", help="Read a Mist's content and metadata from MuseHub.", description=( "Fetch full Mist content and metadata by ID.\n\n" "MIST_ID may be the 12-character mist ID or 'owner/ID' form.\n\n" "Agent quickstart:\n" " muse mist read aB3xKq9dPwNm --json\n" " muse mist read gabriel/aB3xKq9dPwNm --json" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) read_p.add_argument("mist_id", metavar="MIST_ID", help="Mist ID or owner/ID.") read_p.add_argument( "--hub", metavar="URL", default=None, help="Override the MuseHub URL.", ) read_p.add_argument( "--json", "-j", action="store_true", dest="json_output", default=False, help="Emit a JSON object to stdout.", ) read_p.set_defaults(func=run_read) # ── fork ────────────────────────────────────────────────────────────────── fork_p = subs.add_parser( "fork", help="Fork a Mist into the caller's namespace.", description=( "Create a copy of MIST_ID in the authenticated user's namespace.\n" "The fork tracks its upstream; you can submit a proposal back.\n\n" "Agent quickstart:\n" " muse mist fork aB3xKq9dPwNm --json" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) fork_p.add_argument("mist_id", metavar="MIST_ID", help="Mist ID or owner/ID to fork.") fork_p.add_argument( "--hub", metavar="URL", default=None, help="Override the MuseHub URL.", ) fork_p.add_argument( "--json", "-j", action="store_true", dest="json_output", default=False, help="Emit a JSON object to stdout.", ) fork_p.set_defaults(func=run_fork) # ── push ────────────────────────────────────────────────────────────────── push_p = subs.add_parser( "push", help="Push a local Mist repo to MuseHub.", description=( "Must be run from inside a Muse repo with domain='mist'.\n\n" "This is the manual workflow: init a mist repo, add your artifact,\n" "commit, then push. For one-shot creation use:\n" " muse mist create --push\n\n" "Agent quickstart:\n" " muse mist push --remote local --branch main" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) push_p.add_argument( "--remote", "-r", metavar="REMOTE", default="local", help="Remote name to push to (default: local).", ) push_p.add_argument( "--branch", "-b", metavar="BRANCH", default="main", help="Branch to push (default: main).", ) push_p.add_argument( "--json", "-j", action="store_true", dest="json_output", default=False, help="Emit a JSON object to stdout.", ) push_p.set_defaults(func=run_push) # ── embed ───────────────────────────────────────────────────────────────── embed_p = subs.add_parser( "embed", help="Generate embed code (iframe, JS, Markdown badge) for a Mist.", description=( "Generate embeddable HTML, JS snippet, and Markdown badge for MIST_ID.\n\n" "Agent quickstart:\n" " muse mist embed aB3xKq9dPwNm --json\n" " muse mist embed gabriel/aB3xKq9dPwNm --width 800 --height 400" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) embed_p.add_argument("mist_id", metavar="MIST_ID", help="Mist ID or owner/ID.") embed_p.add_argument( "--width", type=int, default=600, metavar="N", help="Embed width in pixels (default: 600).", ) embed_p.add_argument( "--height", type=int, default=300, metavar="N", help="Embed height in pixels (default: 300).", ) embed_p.add_argument( "--hub", metavar="URL", default=None, help="Override the MuseHub URL.", ) embed_p.add_argument( "--json", "-j", action="store_true", dest="json_output", default=False, help="Emit a JSON object to stdout.", ) embed_p.set_defaults(func=run_embed) # ── delete ──────────────────────────────────────────────────────────────── delete_p = subs.add_parser( "delete", help="Delete a Mist from MuseHub (owner only).", description=( "Permanently delete MIST_ID and its underlying Muse repo.\n" "This cannot be undone. Only the owner can delete a Mist.\n\n" "Agent quickstart:\n" " muse mist delete aB3xKq9dPwNm --yes --json" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) delete_p.add_argument("mist_id", metavar="MIST_ID", help="Mist ID or owner/ID to delete.") delete_p.add_argument( "--yes", "-y", action="store_true", default=False, help="Skip the confirmation prompt.", ) delete_p.add_argument( "--hub", metavar="URL", default=None, help="Override the MuseHub URL.", ) delete_p.add_argument( "--json", "-j", action="store_true", dest="json_output", default=False, help="Emit a JSON object to stdout.", ) delete_p.set_defaults(func=run_delete) # ── update ──────────────────────────────────────────────────────────────── update_p = subs.add_parser( "update", help="Update a Mist's title, description, visibility, tags, or content.", description=( "Partial update — only provided flags are changed; omitted flags are left\n" "unchanged. Updating --content increments the mist's version counter.\n\n" "Agent quickstart:\n" " muse mist update aB3xKq9dPwNm --title 'Better title' --json\n" " muse mist update aB3xKq9dPwNm --content new_version.py --json" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) update_p.add_argument("mist_id", metavar="MIST_ID", help="Mist ID or owner/ID to update.") update_p.add_argument( "--title", "-t", default=None, metavar="TEXT", help="New human-readable title.", ) update_p.add_argument( "--description", "-d", default=None, metavar="TEXT", help="New Markdown description.", ) update_p.add_argument( "--visibility", metavar="public|secret", default=None, help="New visibility ('public' or 'secret').", ) update_p.add_argument( "--tags", metavar="TAG,...", default=None, help="Comma-separated tag list (replaces all current tags).", ) update_p.add_argument( "--content", metavar="FILE", default=None, help="Path to a file; its UTF-8 contents replace the artifact. Increments version.", ) update_p.add_argument( "--hub", metavar="URL", default=None, help="Override the MuseHub URL.", ) update_p.add_argument( "--json", "-j", action="store_true", dest="json_output", default=False, help="Emit a JSON object to stdout on success.", ) update_p.set_defaults(func=run_update) # ── forks ───────────────────────────────────────────────────────────────── forks_p = subs.add_parser( "forks", help="List the direct forks of a Mist.", description=( "Fetch GET /api/mists/{mist_id}/forks and display each fork.\n\n" "Agent quickstart:\n" " muse mist forks aB3xKq9dPwNm --json" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) forks_p.add_argument("mist_id", metavar="MIST_ID", help="Mist ID or owner/ID.") forks_p.add_argument( "--limit", "-n", type=int, default=20, metavar="N", help="Maximum forks to return (1–100, default 20).", ) forks_p.add_argument( "--hub", metavar="URL", default=None, help="Override the MuseHub URL.", ) forks_p.add_argument( "--json", "-j", action="store_true", dest="json_output", default=False, help="Emit a JSON array to stdout.", ) forks_p.set_defaults(func=run_forks) # ── raw ─────────────────────────────────────────────────────────────────── raw_p = subs.add_parser( "raw", help="Print or save the raw artifact bytes of a Mist.", description=( "Fetches GET /api/mists/{mist_id}/raw and writes to stdout\n" "or to --output FILE.\n\n" "Agent quickstart:\n" " muse mist raw aB3xKq9dPwNm > validate.py\n" " muse mist raw aB3xKq9dPwNm --output local_copy.py" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) raw_p.add_argument("mist_id", metavar="MIST_ID", help="Mist ID or owner/ID.") raw_p.add_argument( "--output", "-o", metavar="FILE", default=None, help="Write artifact bytes to FILE instead of stdout.", ) raw_p.add_argument( "--hub", metavar="URL", default=None, help="Override the MuseHub URL.", ) raw_p.set_defaults(func=run_raw)