"""muse domains — domain plugin dashboard, scaffold wizard, and protocol validator. Output (default — no flags):: ╔══════════════════════════════════════════════════════════════╗ ║ Muse Domain Plugin Dashboard ║ ╚══════════════════════════════════════════════════════════════╝ Registered domains: 2 ───────────────────────────────────────────────────────────── ● code (active repo domain) Module: plugins/code/plugin.py Capabilities: Typed Deltas · Domain Schema · OT Merge Schema: v1.0 · top_level: record · merge_mode: three_way Dimensions: file, symbol, dependency Description: Version source code as structured multidimensio ○ scaffold Module: plugins/scaffold/plugin.py Capabilities: Typed Deltas · Domain Schema · OT Merge · CRDT Schema: v1.0 · top_level: record · merge_mode: three_way Dimensions: primary, metadata Description: Scaffold domain plugin — copy-paste template fo ───────────────────────────────────────────────────────────── To scaffold a new domain: muse domains --new ───────────────────────────────────────────────────────────── Subcommands:: muse domains publish --author --slug ... Publish a domain plugin to the MuseHub marketplace. muse domains info Show full info for a single registered domain. muse domains use Switch the current repository's active domain. muse domains validate [] Verify a domain plugin correctly implements the MuseDomainPlugin protocol. --json emits machine-readable JSON for every subcommand. --new scaffolds a new domain plugin directory from the scaffold template. Name must match ``^[a-z][a-z0-9_-]{0,63}$`` and must not be an existing plugin name. """ from __future__ import annotations import argparse import json import logging import pathlib import re import shutil import sys import urllib.error import urllib.parse import urllib.request from typing import TYPE_CHECKING, Literal, TypedDict from muse.cli.config import get_signing_identity, get_hub_url from muse.core.repo import find_repo_root from muse.core.store import write_text_atomic from muse.domain import CRDTPlugin, MuseDomainPlugin, StructuredMergePlugin from muse.plugins.registry import _REGISTRY from muse.core.validation import sanitize_display logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- _DEFAULT_DOMAIN = "code" _ALLOWED_PUBLISH_SCHEMES = frozenset({"http", "https"}) _MAX_RESPONSE_BYTES = 4 * 1024 * 1024 # 4 MiB — prevent OOM from malicious servers _PUBLISH_TIMEOUT = 15 # seconds _WIDTH = 62 # Domain names must be lowercase alphanumeric with hyphens or underscores. # The anchor prevents path traversal via "../evil". _DOMAIN_NAME_RE = re.compile(r"^[a-z][a-z0-9_-]{0,63}$") # --------------------------------------------------------------------------- # Internal types — capabilities / JSON output # --------------------------------------------------------------------------- _CapabilityLabel = Literal["Typed Deltas", "Domain Schema", "OT Merge", "CRDT"] class _DimensionDef(TypedDict): """One semantic dimension exported by a domain plugin.""" name: str description: str class _SchemaInfoJson(TypedDict): """Schema block embedded in a domain's JSON entry. Absent when the plugin does not implement ``schema()`` — agents should check ``"schema" in entry`` before accessing. """ schema_version: str merge_mode: str description: str dimensions: list[_DimensionDef] class _DomainEntryJsonBase(TypedDict): """Required keys present on every domain JSON entry.""" domain: str module_path: str capabilities: list[str] active: bool class _DomainEntryJson(_DomainEntryJsonBase, total=False): """Full per-domain JSON entry — ``schema`` is omitted when unavailable.""" schema: _SchemaInfoJson class _ScaffoldJson(TypedDict): """Structured result from ``muse domains --new --json``.""" name: str class_name: str path: str status: str class _UseJson(TypedDict): """Result from ``muse domains use --json``.""" domain: str repo: str status: str class _ValidateCheckJson(TypedDict): """One protocol compliance check from ``muse domains validate``.""" name: str ok: bool detail: str class _ValidateJson(TypedDict): """Full result from ``muse domains validate [] --json``.""" domain: str ok: bool checks: list[_ValidateCheckJson] # --------------------------------------------------------------------------- # Publish-specific types (MuseHub wire format) # --------------------------------------------------------------------------- class _Capabilities(TypedDict, total=False): """Capability manifest sent to MuseHub on domain publish. ``total=False`` because MuseHub accepts partial manifests and fills defaults. When derived from a plugin ``schema()``, ``dimensions`` and ``merge_semantics`` are always populated. """ dimensions: list[_DimensionDef] artifact_types: list[str] merge_semantics: str supported_commands: list[str] class _PublishPayload(TypedDict): """Wire payload for ``POST /api/v1/domains`` on MuseHub.""" author_slug: str slug: str display_name: str description: str capabilities: _Capabilities viewer_type: str version: str class _PublishResponse(TypedDict): """Parsed response body from ``POST /api/v1/domains``. All keys are guaranteed present — ``_post_json`` normalises missing keys to empty strings so callers never hit ``KeyError``. """ domain_id: str scoped_id: str manifest_hash: str def _plugin_module_path(name: str) -> str: """Return the display-friendly module path for a plugin. Args: name: Domain name string (key in the registry). Returns: Path string like ``plugins/code/plugin.py``. """ return f"plugins/{name}/plugin.py" # --------------------------------------------------------------------------- # Helpers — repository context # --------------------------------------------------------------------------- def _active_domain(ctx: pathlib.Path | None) -> str | None: """Return the domain name of the repository at *ctx*, or ``None``. Returns ``None`` — not a default domain name — when no repo is found so callers can distinguish "inside a repo with an explicit domain" from "not inside any repo at all". Args: ctx: Repo root path or ``None`` when not inside a repo. Returns: Domain name string or ``None``. """ if ctx is None: return None repo_json = ctx / ".muse" / "repo.json" if not repo_json.exists(): return None try: data = json.loads(repo_json.read_text(encoding="utf-8")) domain = data.get("domain") return str(domain) if domain else _DEFAULT_DOMAIN except (OSError, json.JSONDecodeError): return None def _find_repo_root() -> pathlib.Path | None: """Find the current repository root. Returns: The repo root :class:`pathlib.Path`, or ``None`` when not inside a repo. """ return find_repo_root() # --------------------------------------------------------------------------- # Helpers — validation # --------------------------------------------------------------------------- def _validate_domain_name(name: str) -> None: """Raise ``SystemExit(1)`` if *name* is not a safe domain name. Rules: - Must match ``^[a-z][a-z0-9_-]{0,63}$`` — blocks path traversal sequences such as ``../evil`` and shell-special characters. - Must not be ``"scaffold"`` — that is the built-in template directory. Args: name: The raw domain name supplied by the user. """ if not _DOMAIN_NAME_RE.match(name): print( f"❌ Invalid domain name {sanitize_display(name)!r}.\n" " Names must start with a lowercase letter, contain only " "lowercase letters, digits, hyphens, or underscores, and be " "at most 64 characters.", file=sys.stderr, ) raise SystemExit(1) if name == "scaffold": print( "❌ 'scaffold' is the built-in template — choose a different name.", file=sys.stderr, ) raise SystemExit(1) def _validate_publish_url(url: str) -> None: """Raise ``SystemExit(1)`` if *url* has a non-HTTP/HTTPS scheme. Prevents SSRF via ``file://``, ``ftp://``, ``javascript:``, etc. Args: url: The resolved hub URL to validate. """ parsed = urllib.parse.urlparse(url) if parsed.scheme not in _ALLOWED_PUBLISH_SCHEMES: print( f"❌ Blocked hub URL scheme {sanitize_display(parsed.scheme)!r} — " "only http:// and https:// are allowed.", file=sys.stderr, ) raise SystemExit(1) # --------------------------------------------------------------------------- # JSON emitter # --------------------------------------------------------------------------- def _build_entry(domain_name: str, plugin: MuseDomainPlugin, active_domain: str | None) -> _DomainEntryJson: """Build a single ``_DomainEntryJson`` for *domain_name*. Calls ``plugin.schema()`` once and reuses the result for both the capabilities check and the schema block — avoids the double-call that was present in earlier code. Args: domain_name: Registry key. plugin: Registered plugin instance. active_domain: The current repo's domain (for the ``active`` flag). Returns: A fully populated ``_DomainEntryJson`` with ``schema`` omitted when the plugin raises ``NotImplementedError``. """ caps: list[_CapabilityLabel] = ["Typed Deltas"] schema_result: _SchemaInfoJson | None = None try: s = plugin.schema() caps.append("Domain Schema") if isinstance(plugin, StructuredMergePlugin): caps.append("OT Merge") if isinstance(plugin, CRDTPlugin): caps.append("CRDT") schema_result = _SchemaInfoJson( schema_version=str(s["schema_version"]), merge_mode=s["merge_mode"], description=s["description"], dimensions=[ _DimensionDef(name=d["name"], description=d["description"]) for d in s["dimensions"] ], ) except NotImplementedError: pass entry = _DomainEntryJson( domain=domain_name, module_path=_plugin_module_path(domain_name), capabilities=list(caps), active=domain_name == active_domain, ) if schema_result is not None: entry["schema"] = schema_result return entry def _emit_json(active_domain: str | None) -> None: """Print all registered domains and their capabilities as JSON to stdout. Output schema (one element per registered domain):: [ { "domain": "code", "module_path": "plugins/code/plugin.py", "capabilities": ["Typed Deltas", "Domain Schema", "OT Merge"], "active": true, "schema": { "schema_version": "1.0", "merge_mode": "three_way", "description": "...", "dimensions": [{"name": "file", "description": "..."}, ...] } }, ... ] ``schema`` is absent when the plugin does not implement ``schema()``. ``active`` is a boolean — ``true`` when this domain matches the current repository's configured domain. Args: active_domain: The domain of the current repo, or ``None``. """ result: list[_DomainEntryJson] = [ _build_entry(name, plugin, active_domain) for name, plugin in sorted(_REGISTRY.items()) ] print(json.dumps(result, indent=2)) # --------------------------------------------------------------------------- # Human-readable dashboard # --------------------------------------------------------------------------- def _box_line(text: str) -> str: """Center *text* inside a box line of width ``_WIDTH``.""" inner = _WIDTH - 2 padded = text.center(inner) return f"║{padded}║" def _hr() -> str: """Return a horizontal rule of width ``_WIDTH``.""" return "─" * _WIDTH def _print_dashboard(active_domain: str | None) -> None: """Print the human-readable domain dashboard to stdout. Args: active_domain: Domain of the current repo (highlighted with ●), or ``None``. """ print("╔" + "═" * (_WIDTH - 2) + "╗") print(_box_line("Muse Domain Plugin Dashboard")) print("╚" + "═" * (_WIDTH - 2) + "╝") print("") count = len(_REGISTRY) print(f"Registered domains: {count}") print(_hr()) for domain_name, plugin in sorted(_REGISTRY.items()): is_active = domain_name == active_domain bullet = "●" if is_active else "○" safe_name = sanitize_display(domain_name) print("") active_suffix = " (active repo domain)" if is_active else "" print(f" {bullet} {safe_name}{active_suffix}") print(f" Module: {_plugin_module_path(domain_name)}") # Call schema() once; reuse for capabilities and schema display. caps: list[_CapabilityLabel] = ["Typed Deltas"] try: s = plugin.schema() caps.append("Domain Schema") if isinstance(plugin, StructuredMergePlugin): caps.append("OT Merge") if isinstance(plugin, CRDTPlugin): caps.append("CRDT") print(f" Capabilities: {' · '.join(caps)}") dim_names = [d["name"] for d in s["dimensions"]] top_kind = s["top_level"]["kind"] print( f" Schema: v{s['schema_version']} · " f"top_level: {top_kind} · merge_mode: {s['merge_mode']}" ) print(f" Dimensions: {', '.join(dim_names)}") print(f" Description: {sanitize_display(s['description'][:55])}") except NotImplementedError: print(f" Capabilities: {' · '.join(caps)}") print(" Schema: (not declared)") print("") print(_hr()) print("To scaffold a new domain:") print(" muse domains --new ") print("To inspect a specific domain:") print(" muse domains info ") print("To switch the current repo's domain:") print(" muse domains use ") print("To validate protocol compliance:") print(" muse domains validate []") print("To see machine-readable output:") print(" muse domains --json") print("See docs/guide/plugin-authoring-guide.md for the full walkthrough.") print(_hr()) # --------------------------------------------------------------------------- # Scaffold wizard # --------------------------------------------------------------------------- def _scaffold_new_domain(name: str, as_json: bool) -> None: """Create a new plugin directory by copying the scaffold template. Copies ``muse/plugins/scaffold/`` to ``muse/plugins//``, then renames ``ScaffoldPlugin`` to ``Plugin`` in the source files. ``__pycache__`` and bytecode are excluded from the copy. Security: *name* must pass ``_validate_domain_name()`` before this function is called — the function asserts the constraint and exits early if it detects a path traversal attempt or reserved name. Args: name: The new domain name (validated before this call). as_json: When ``True`` emit a ``_ScaffoldJson`` dict to stdout. """ scaffold_src = pathlib.Path(__file__).parents[2] / "plugins" / "scaffold" dest = pathlib.Path(__file__).parents[2] / "plugins" / name if dest.exists(): print( f"❌ Plugin directory already exists: {sanitize_display(str(dest))}", file=sys.stderr, ) raise SystemExit(1) if not scaffold_src.exists(): print( "❌ Scaffold source not found. Make sure muse/plugins/scaffold/ exists.", file=sys.stderr, ) raise SystemExit(1) shutil.copytree( str(scaffold_src), str(dest), ignore=shutil.ignore_patterns("__pycache__", "*.pyc", "*.pyo"), ) class_name = "".join(part.capitalize() for part in name.split("_")) + "Plugin" for py_file in dest.glob("*.py"): text = py_file.read_text(encoding="utf-8") text = text.replace("ScaffoldPlugin", class_name) text = text.replace('_DOMAIN_NAME = "scaffold"', f'_DOMAIN_NAME = "{name}"') text = text.replace( "Scaffold domain plugin — copy-paste template for a new Muse domain.", f"{class_name} — Muse domain plugin for the {name!r} domain.", ) py_file.write_text(text, encoding="utf-8") if as_json: result = _ScaffoldJson( name=name, class_name=class_name, path=f"muse/plugins/{name}/", status="ok", ) print(json.dumps(result, indent=2)) return print(f"✅ Scaffolded new domain plugin: muse/plugins/{sanitize_display(name)}/") print(f" Class name: {sanitize_display(class_name)}") print("") print("Next steps:") print(f" 1. Implement every NotImplementedError in muse/plugins/{sanitize_display(name)}/plugin.py") print(" 2. Register the plugin in muse/plugins/registry.py:") print(f' from muse.plugins.{sanitize_display(name)}.plugin import {sanitize_display(class_name)}') print(f' _REGISTRY["{sanitize_display(name)}"] = {sanitize_display(class_name)}()') print(f' 3. muse init --domain {sanitize_display(name)}') print(" 4. See docs/guide/plugin-authoring-guide.md for the full walkthrough") # --------------------------------------------------------------------------- # Publish subcommand # --------------------------------------------------------------------------- def _post_json(url: str, payload: _PublishPayload, signing: "SigningIdentity") -> _PublishResponse: """HTTP POST *payload* as JSON to *url* authenticated with MSign. Uses :mod:`urllib.request` (no third-party dependencies) with a ``Content-Type: application/json`` body and ``Authorization: MSign`` header. The timeout is :data:`_PUBLISH_TIMEOUT` seconds and the response body is capped at :data:`_MAX_RESPONSE_BYTES` to prevent OOM. The caller is responsible for validating *url*'s scheme via ``_validate_publish_url`` before invoking this function. Args: url: Full endpoint URL with a validated http/https scheme. payload: Typed publish payload — serialised verbatim to JSON. signing: :class:`~muse.core.transport.SigningIdentity` from identity.toml. Returns: Parsed ``_PublishResponse`` with ``domain_id``, ``scoped_id``, and ``manifest_hash`` from the server. Missing keys are normalised to empty strings. Raises: urllib.error.HTTPError: on non-2xx HTTP responses. urllib.error.URLError: on DNS/connection failure. ValueError: when the response body is not a JSON object. """ from muse.core.transport import SigningIdentity, build_msign_header body = json.dumps(payload).encode() req = urllib.request.Request( url, data=body, headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": build_msign_header(signing, "POST", url, body), }, method="POST", ) with urllib.request.urlopen(req, timeout=_PUBLISH_TIMEOUT) as resp: # noqa: S310 raw = resp.read(_MAX_RESPONSE_BYTES).decode(errors="replace") parsed = json.loads(raw) if not isinstance(parsed, dict): raise ValueError(f"Expected JSON object from server, got: {type(parsed).__name__}") return _PublishResponse( domain_id=str(parsed.get("domain_id") or ""), scoped_id=str(parsed.get("scoped_id") or ""), manifest_hash=str(parsed.get("manifest_hash") or ""), ) def run_publish(args: argparse.Namespace) -> None: """Publish a Muse domain plugin to the MuseHub marketplace. Registers ``@{author}/{slug}`` so agents and users can discover and install the domain via ``musehub_list_domains`` and ``muse domains``. Capabilities are read from the active domain plugin's ``schema()`` when ``--capabilities`` is omitted — so you can run this command from inside a repo that uses the domain you want to publish. Security: - The hub URL scheme is validated before any network request — ``file://``, ``ftp://``, and similar non-HTTP/HTTPS schemes are rejected (SSRF guard). - The response body is capped at 4 MiB to prevent OOM from malicious servers. - All server-returned values are passed through ``sanitize_display()`` before appearing in human-readable output. - The signing identity is never logged or printed. JSON output fields (``--json`` / ``-j``) ----------------------------------------- ``domain_id`` Hub-assigned opaque identifier for the registered domain. ``scoped_id`` Canonical scoped name in ``@{author}/{slug}`` form, e.g. ``"@gabriel/genomics"``. Use this value in subsequent ``musehub_get_domain`` or ``musehub_create_repo`` calls. ``manifest_hash`` Content hash of the published capability manifest, e.g. ``"sha256:abc123"``. Useful for auditing and reproducibility. Exit codes ---------- 0 — domain published successfully 1 — auth token missing, invalid hub URL scheme, bad ``--capabilities`` JSON, plugin schema unavailable, HTTP error, or connection failure 2 — not inside a Muse repository (only when ``--hub`` is not provided and no config.toml hub URL is set) Example:: muse domains publish \\ --author gabriel --slug genomics \\ --name "Genomics" \\ --description "Version DNA sequences as multidimensional state" \\ --viewer-type genome muse domains publish --author gabriel --slug spatial \\ --name "Spatial 3D" \\ --description "Version 3-D scenes as structured multidimensional commits" \\ --viewer-type spatial \\ --capabilities '{"dimensions":[{"name":"geometry","description":"Mesh data"}],...}' """ author_slug: str = args.author_slug slug: str = args.slug display_name: str = args.display_name description: str = args.description viewer_type: str = args.viewer_type version: str = args.version capabilities_json: str | None = args.capabilities_json hub_url: str | None = args.hub_url as_json: bool = args.as_json # ── Resolve hub URL and validate scheme ──────────────────────────────────── repo_root = find_repo_root() resolved_hub = hub_url or get_hub_url(repo_root) or "https://musehub.ai" resolved_hub = resolved_hub.rstrip("/") _validate_publish_url(resolved_hub) token = get_signing_identity(repo_root) if not token: print( "❌ No signing identity found. Run:\n" " muse auth keygen --hub \n" " muse auth register --hub --handle ", file=sys.stderr, ) raise SystemExit(1) # ── Build capabilities manifest ──────────────────────────────────────────── capabilities: _Capabilities if capabilities_json is not None: try: raw_caps = json.loads(capabilities_json) if not isinstance(raw_caps, dict): raise ValueError("capabilities JSON must be an object") capabilities = _Capabilities( dimensions=[ _DimensionDef(name=str(d.get("name", "")), description=str(d.get("description", ""))) for d in raw_caps.get("dimensions", []) if isinstance(d, dict) ], artifact_types=[str(a) for a in raw_caps.get("artifact_types", []) if isinstance(a, str)], merge_semantics=str(raw_caps.get("merge_semantics", "three_way")), supported_commands=[str(c) for c in raw_caps.get("supported_commands", []) if isinstance(c, str)], ) except (json.JSONDecodeError, ValueError) as exc: print(f"❌ --capabilities is not valid JSON: {exc}", file=sys.stderr) raise SystemExit(1) from exc else: # Derive from the active domain plugin schema when available. active_domain_name: str | None = None if repo_root is not None: repo_json = repo_root / ".muse" / "repo.json" try: active_domain_name = json.loads(repo_json.read_text(encoding="utf-8")).get("domain") except (OSError, json.JSONDecodeError): pass plugin = _REGISTRY.get(active_domain_name or "") if active_domain_name else None capabilities_ok = False if plugin is not None: try: schema = plugin.schema() capabilities = _Capabilities( dimensions=[ _DimensionDef(name=d["name"], description=d["description"]) for d in schema["dimensions"] ], artifact_types=[], merge_semantics=schema["merge_mode"], supported_commands=["commit", "diff", "merge", "log", "status"], ) capabilities_ok = True except NotImplementedError: capabilities = _Capabilities() else: capabilities = _Capabilities() if not capabilities_ok: print( "⚠️ Could not derive capabilities from active plugin. " "Provide --capabilities '' to set them explicitly.", file=sys.stderr, ) print( " Required keys: dimensions, artifact_types, merge_semantics, supported_commands", file=sys.stderr, ) raise SystemExit(1) # ── POST to MuseHub ──────────────────────────────────────────────────────── endpoint = f"{resolved_hub}/api/v1/domains" payload = _PublishPayload( author_slug=author_slug, slug=slug, display_name=display_name, description=description, capabilities=capabilities, viewer_type=viewer_type, version=version, ) try: result = _post_json(endpoint, payload, token) except urllib.error.HTTPError as exc: body = exc.read().decode(errors="replace") if exc.code == 409: print( f"❌ Domain '@{sanitize_display(author_slug)}/{sanitize_display(slug)}' " "is already registered. Use a different slug or bump the version.", file=sys.stderr, ) elif exc.code == 401: print("❌ Authentication failed — is your MuseHub token valid?", file=sys.stderr) else: print(f"❌ MuseHub returned HTTP {exc.code}: {sanitize_display(body[:200])}", file=sys.stderr) raise SystemExit(1) from exc except urllib.error.URLError as exc: print( f"❌ Could not reach MuseHub at {sanitize_display(resolved_hub)}: " f"{sanitize_display(str(exc.reason))}", file=sys.stderr, ) raise SystemExit(1) from exc except ValueError as exc: print(f"❌ Unexpected response from MuseHub: {sanitize_display(str(exc))}", file=sys.stderr) raise SystemExit(1) from exc # ── Emit result ──────────────────────────────────────────────────────────── if as_json: print(json.dumps(result)) return scoped_id = sanitize_display(result.get("scoped_id") or f"@{author_slug}/{slug}") manifest_hash = sanitize_display(result.get("manifest_hash") or "") safe_hub = sanitize_display(resolved_hub) safe_author = sanitize_display(author_slug) safe_slug = sanitize_display(slug) print(f"✅ Domain published: {scoped_id}") print(f" manifest_hash: {manifest_hash}") print(f" Discoverable at: {safe_hub}/domains/@{safe_author}/{safe_slug}") print("") print("Agents can now use it:") print(f' musehub_get_domain(scoped_id="{scoped_id}")') print(f' musehub_create_repo(domain="{scoped_id}", ...)') # --------------------------------------------------------------------------- # Info subcommand — targeted per-domain query (agent-native) # --------------------------------------------------------------------------- def run_info(args: argparse.Namespace) -> None: """Show full information for a single registered domain plugin. Looks up *name* in the in-process plugin registry, builds a capability summary by calling ``plugin.schema()`` once (if implemented), and emits either a human-readable table or a JSON object. Security: the domain name supplied on the command line is passed through ``sanitize_display()`` before appearing in error messages. In text mode, all plugin-sourced string fields (module path, schema version, merge mode, dimension names, description) are sanitized before output so that a malicious third-party plugin cannot inject terminal escape sequences via its ``schema()`` return value. JSON output fields (``--json`` / ``-j``) ----------------------------------------- ``domain`` Registry key (the name passed on the command line). ``module_path`` Relative filesystem path to the plugin's Python module. ``capabilities`` Array of capability label strings present on this plugin, e.g. ``["Typed Deltas", "Domain Schema", "OT Merge", "CRDT"]``. ``active`` ``true`` when this domain matches the current repo's configured domain. ``schema`` *(optional — absent when plugin does not implement* ``schema()`` *)* Object with: ``schema_version`` semver string declared by the plugin. ``merge_mode`` merge strategy, e.g. ``"structured"``. ``description`` human-readable summary of the domain. ``dimensions`` array of ``{"name": str, "description": str}`` objects. Exit codes ---------- 0 — domain found and info emitted 1 — domain name is not registered """ name: str = args.info_name as_json: bool = args.as_json plugin = _REGISTRY.get(name) if plugin is None: known = ", ".join(sorted(_REGISTRY)) print( f"❌ Domain {sanitize_display(name)!r} is not registered. " f"Known domains: {sanitize_display(known)}", file=sys.stderr, ) raise SystemExit(1) active_domain = _active_domain(_find_repo_root()) entry = _build_entry(name, plugin, active_domain) if as_json: print(json.dumps(entry)) return safe_name = sanitize_display(name) is_active = entry["active"] active_suffix = " (active repo domain)" if is_active else "" print(f"{'●' if is_active else '○'} {safe_name}{active_suffix}") print(f" Module: {sanitize_display(entry['module_path'])}") print(f" Capabilities: {' · '.join(sanitize_display(c) for c in entry['capabilities'])}") if "schema" in entry: s = entry["schema"] dim_names = [sanitize_display(d["name"]) for d in s["dimensions"]] print( f" Schema: v{sanitize_display(str(s['schema_version']))} · " f"merge_mode: {sanitize_display(s['merge_mode'])}" ) print(f" Dimensions: {', '.join(dim_names)}") print(f" Description: {sanitize_display(s['description'])}") else: print(" Schema: (not declared)") # --------------------------------------------------------------------------- # Use subcommand — switch the active domain for the current repo # --------------------------------------------------------------------------- def run_use(args: argparse.Namespace) -> None: """Switch the current repository's active domain. Writes the new domain name to ``.muse/repo.json`` atomically using ``write_text_atomic`` (mkstemp → fsync → rename) so a crash during the update cannot corrupt the file. All other fields in ``repo.json`` are preserved verbatim. The domain must be registered in the plugin registry — you cannot switch to a domain that Muse does not know how to handle. The operation is idempotent: switching to the already-active domain exits 0 without a write. Security: the domain name supplied on the command line is validated against the in-process plugin registry before it is written to ``repo.json``, so only known registered names can be stored. ``write_text_atomic`` rejects symlinked parent directories to prevent symlink-swap attacks. The domain name and repo path are passed through ``sanitize_display()`` before appearing in text output. JSON output fields (``--json`` / ``-j``) ----------------------------------------- ``domain`` The domain name that is now active (echoed from the argument). ``repo`` Absolute path to the ``.muse/`` directory that was updated. ``status`` Always ``"switched"`` on success. Exit codes ---------- 0 — domain switched (or already active — idempotent) 1 — not inside a Muse repository, domain not registered, or repo.json could not be read """ name: str = args.use_name as_json: bool = args.as_json ctx = _find_repo_root() if ctx is None: print("❌ Not inside a Muse repository.", file=sys.stderr) raise SystemExit(1) if name not in _REGISTRY: known = ", ".join(sorted(_REGISTRY)) print( f"❌ Domain {sanitize_display(name)!r} is not registered. " f"Known domains: {sanitize_display(known)}", file=sys.stderr, ) raise SystemExit(1) muse_dir = ctx / ".muse" repo_json_path = muse_dir / "repo.json" try: data = json.loads(repo_json_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: print(f"❌ Could not read repo.json: {exc}", file=sys.stderr) raise SystemExit(1) from exc data["domain"] = name write_text_atomic(repo_json_path, json.dumps(data, indent=2) + "\n") muse_dir_str = sanitize_display(str(muse_dir)) if as_json: result = _UseJson(domain=name, repo=str(muse_dir), status="switched") print(json.dumps(result)) return print(f"✅ Active domain switched to {sanitize_display(name)!r}") print(f" Repo: {muse_dir_str}") # --------------------------------------------------------------------------- # Validate subcommand — protocol compliance checker # --------------------------------------------------------------------------- def _check_method(plugin: MuseDomainPlugin, method: str) -> _ValidateCheckJson: """Return a validate check for whether *plugin* has a callable *method*.""" has_it = callable(getattr(plugin, method, None)) return _ValidateCheckJson( name=f"has_method:{method}", ok=has_it, detail="present" if has_it else f"missing or not callable: {method}", ) def _run_validate_plugin(name: str, plugin: MuseDomainPlugin, active_domain: str | None) -> _ValidateJson: """Run all protocol compliance checks for *plugin* and return the result. Checks performed: 1. Required ``MuseDomainPlugin`` methods exist and are callable. 2. ``schema()`` returns without raising (``NotImplementedError`` is noted as a missing capability, not a hard failure). 3. ``StructuredMergePlugin.merge_structured`` exists when the plugin advertises OT Merge capability. 4. ``CRDTPlugin.merge_crdt`` exists when the plugin advertises CRDT. Args: name: Registry key for the plugin. plugin: Plugin instance to check. active_domain: Current repo's domain (for context display only). Returns: A ``_ValidateJson`` with per-check details and an overall ``ok`` flag. """ checks: list[_ValidateCheckJson] = [] # Required protocol methods (actual MuseDomainPlugin protocol method names) for method in ("snapshot", "diff", "merge", "drift", "apply"): checks.append(_check_method(plugin, method)) # schema() — optional but strongly recommended; catches AttributeError for # plugins that don't even define the attribute. try: plugin.schema() checks.append(_ValidateCheckJson(name="schema()", ok=True, detail="implemented")) except NotImplementedError: checks.append(_ValidateCheckJson( name="schema()", ok=False, detail="raises NotImplementedError — Domain Schema capability unavailable", )) except AttributeError: checks.append(_ValidateCheckJson( name="schema()", ok=False, detail="attribute missing — Domain Schema capability unavailable", )) # Optional protocol extensions — only checked when the plugin claims them if isinstance(plugin, StructuredMergePlugin): checks.append(_check_method(plugin, "merge_ops")) if isinstance(plugin, CRDTPlugin): checks.append(_check_method(plugin, "join")) all_ok = all(c["ok"] for c in checks) return _ValidateJson(domain=name, ok=all_ok, checks=checks) def run_validate(args: argparse.Namespace) -> None: """Verify a domain plugin correctly implements the MuseDomainPlugin protocol. Checks that required ``MuseDomainPlugin`` methods are callable (``snapshot``, ``diff``, ``merge``, ``drift``, ``apply``), that ``schema()`` does not raise unexpectedly, and that optional capability interfaces (``StructuredMergePlugin``, ``CRDTPlugin``) are correctly implemented when the plugin claims them. When *name* is omitted the active repo's domain is validated; when not inside a repo, all registered domains are validated. With ``--json`` a single-domain result is emitted as an object; a multi-domain result is emitted as an array. Security: the *name* argument is validated against the plugin registry before any checks are performed — unregistered names are rejected with exit code 1 and never reach the check logic. All text-mode output passes through ``sanitize_display()`` so domain names or check details containing ANSI escapes cannot corrupt the terminal. JSON output fields (``--json`` / ``-j``) ----------------------------------------- ``domain`` Registry key of the validated plugin. ``ok`` ``true`` when every check passed; ``false`` otherwise. ``checks`` List of per-check objects, each with: ``name`` Check identifier (e.g. ``"has_method:snapshot"``). ``ok`` ``true`` if the check passed. ``detail`` Human-readable result (e.g. ``"present"`` or reason). Exit codes ---------- 0 — all checks passed (or all domains passed when validating all) 1 — one or more checks failed, specified domain not registered, or no domains are registered """ name: str | None = args.validate_name as_json: bool = args.as_json active_domain = _active_domain(_find_repo_root()) # Resolve which domain(s) to validate if name is not None: if name not in _REGISTRY: known = ", ".join(sorted(_REGISTRY)) print( f"❌ Domain {sanitize_display(name)!r} is not registered. " f"Known domains: {sanitize_display(known)}", file=sys.stderr, ) raise SystemExit(1) targets: list[tuple[str, MuseDomainPlugin]] = [(name, _REGISTRY[name])] elif active_domain is not None and active_domain in _REGISTRY: targets = [(active_domain, _REGISTRY[active_domain])] else: targets = list(sorted(_REGISTRY.items())) results = [_run_validate_plugin(n, p, active_domain) for n, p in targets] all_ok = all(r["ok"] for r in results) if as_json: output: _ValidateJson | list[_ValidateJson] = results[0] if len(results) == 1 else results print(json.dumps(output)) else: for r in results: icon = "✅" if r["ok"] else "❌" print(f"{icon} {sanitize_display(r['domain'])}") for c in r["checks"]: check_icon = " ✓" if c["ok"] else " ✗" print(f"{check_icon} {c['name']}: {sanitize_display(c['detail'])}") print("") if not all_ok: raise SystemExit(1) # --------------------------------------------------------------------------- # CLI wiring # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``domains`` command and its subcommands.""" parser = subparsers.add_parser( "domains", help="Domain plugin dashboard — list registered domains and their capabilities.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--new", default=None, metavar="NAME", help="Scaffold a new domain plugin with the given name.", ) parser.add_argument( "--json", action="store_true", dest="as_json", help="Emit domain registry (or scaffold result) as JSON.", ) parser.set_defaults(func=run) subs = parser.add_subparsers(dest="subcommand", metavar="SUBCOMMAND") # ── info ─────────────────────────────────────────────────────────────────── info_p = subs.add_parser( "info", help="Show full information for a single registered domain.", description=( "Look up NAME in the in-process plugin registry and print its\n" "capability labels, module path, and (when implemented) full schema\n" "including merge mode, dimensions, and description.\n\n" "Agent quickstart\n" "----------------\n" " muse domains info code --json\n" " muse domains info code -j\n" " muse domains info code -j | jq .capabilities\n" " muse domains info code -j | jq .schema.dimensions\n\n" "JSON output schema\n" "------------------\n" ' {"domain": "", "module_path": "",\n' ' "capabilities": ["Typed Deltas", ...], "active": ,\n' ' "schema": {"schema_version": "", "merge_mode": "",\n' ' "description": "",\n' ' "dimensions": [{"name": "", "description": ""}, ...]}}\n\n' " Note: 'schema' key is absent when the plugin does not implement schema().\n\n" "Exit codes\n" "----------\n" " 0 — domain found and info emitted\n" " 1 — domain name is not registered\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) info_p.add_argument("info_name", metavar="NAME", help="Domain name to inspect.") info_p.add_argument("--json", "-j", action="store_true", dest="as_json", help="Emit as JSON.") info_p.set_defaults(func=run_info) # ── publish ──────────────────────────────────────────────────────────────── publish_p = subs.add_parser( "publish", help="Publish a Muse domain plugin to the MuseHub marketplace.", description=( "Register ``@{author}/{slug}`` on MuseHub so agents and users can\n" "discover and install the domain via ``musehub_list_domains`` and\n" "``muse domains``.\n\n" "Capabilities are derived from the active repo's domain plugin\n" "``schema()`` when ``--capabilities`` is omitted. Pass an explicit\n" "JSON string to override or when not inside a repo.\n\n" "Agent quickstart\n" "----------------\n" " muse domains publish --author gabriel --slug genomics \\\n" " --name Genomics --description '...' --viewer-type genome\n" " muse domains publish ... --json\n" " muse domains publish ... -j | jq .scoped_id\n\n" "JSON output schema\n" "------------------\n" ' {"domain_id": "", "scoped_id": "@/",\n' ' "manifest_hash": ""}\n\n' "Exit codes\n" "----------\n" " 0 — domain published successfully\n" " 1 — auth missing, bad URL scheme, bad --capabilities JSON,\n" " plugin schema unavailable, HTTP error, or connection failure\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) publish_p.add_argument( "--author", required=True, metavar="SLUG", dest="author_slug", help="Your MuseHub username (owner of the domain, e.g. 'gabriel').", ) publish_p.add_argument( "--slug", required=True, metavar="SLUG", help="URL-safe domain name (e.g. 'genomics', 'spatial-3d').", ) publish_p.add_argument( "--name", required=True, metavar="NAME", dest="display_name", help="Human-readable marketplace name (e.g. 'Genomics').", ) publish_p.add_argument( "--description", required=True, metavar="TEXT", help="What this domain models and why it benefits from semantic VCS.", ) publish_p.add_argument( "--viewer-type", required=True, metavar="TYPE", dest="viewer_type", help="Primary viewer identifier (e.g. 'midi', 'code', 'spatial', 'genome').", ) publish_p.add_argument( "--version", default="0.1.0", metavar="SEMVER", help="Semver release string (default: 0.1.0).", ) publish_p.add_argument( "--capabilities", default=None, metavar="JSON", dest="capabilities_json", help=( "Full capabilities manifest as a JSON string. " "Required keys: dimensions, artifact_types, merge_semantics, supported_commands. " "When omitted the active repo's domain plugin schema is used." ), ) publish_p.add_argument( "--hub", default=None, metavar="URL", dest="hub_url", help="Override the MuseHub base URL (default: read from .muse/config.toml).", ) publish_p.add_argument( "--json", "-j", action="store_true", dest="as_json", help="Emit result as JSON.", ) publish_p.set_defaults(func=run_publish) # ── use ──────────────────────────────────────────────────────────────────── use_p = subs.add_parser( "use", help="Switch the current repository's active domain.", description=( "Write NAME as the active domain in ``.muse/repo.json`` using an\n" "atomic rename so a crash cannot corrupt the file. All other\n" "fields in repo.json are preserved. The operation is idempotent —\n" "switching to the already-active domain exits 0 without a write.\n\n" "NAME must be a domain registered in the Muse plugin registry.\n" "Run ``muse domains`` to list all registered domains.\n\n" "Agent quickstart\n" "----------------\n" " muse domains use code --json\n" " muse domains use code -j\n" " muse domains use code -j | jq .status\n\n" "JSON output schema\n" "------------------\n" ' {"domain": "", "repo": "",\n' ' "status": "switched"}\n\n' "Exit codes\n" "----------\n" " 0 — domain switched (or already active)\n" " 1 — not inside a Muse repository, domain not registered,\n" " or repo.json could not be read\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) use_p.add_argument("use_name", metavar="NAME", help="Domain name to activate.") use_p.add_argument("--json", "-j", action="store_true", dest="as_json", help="Emit result as JSON.") use_p.set_defaults(func=run_use) # ── validate ─────────────────────────────────────────────────────────────── validate_p = subs.add_parser( "validate", help="Verify a domain plugin correctly implements the MuseDomainPlugin protocol.", description=( "Run protocol compliance checks on a registered domain plugin.\n" "Checks required methods (snapshot, diff, merge, drift, apply),\n" "schema() availability, and optional capability interfaces.\n\n" "When NAME is omitted, validates the active repo's domain; when\n" "not inside a repo, all registered domains are validated.\n\n" "Agent quickstart\n" "----------------\n" " muse domains validate code --json\n" " muse domains validate code -j\n" " muse domains validate -j # active repo domain\n" " muse domains validate -j | jq .ok\n" " muse domains validate -j | jq '.checks[] | select(.ok == false)'\n\n" "JSON output schema\n" "------------------\n" " Single domain:\n" ' {"domain": "", "ok": ,\n' ' "checks": [{"name": "", "ok": , "detail": ""}, ...]}\n\n' " Multiple domains (no NAME, not in a repo):\n" " [, ...]\n\n" "Exit codes\n" "----------\n" " 0 — all checks passed\n" " 1 — one or more checks failed or domain not registered\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) validate_p.add_argument( "validate_name", nargs="?", default=None, metavar="NAME", help="Domain to validate. Defaults to the active repo domain (or all domains if not in a repo).", ) validate_p.add_argument("--json", "-j", action="store_true", dest="as_json", help="Emit result as JSON.") validate_p.set_defaults(func=run_validate) def run(args: argparse.Namespace) -> None: """Domain plugin dashboard — list registered domains and their capabilities. Without flags: prints a human-readable table of all registered domains, their capability levels (Typed Deltas / Domain Schema / OT Merge / CRDT), and their declared schemas. Flags:: --new Scaffold a new domain plugin directory. --json Machine-readable JSON output (also works with --new). Subcommands:: muse domains publish Push a plugin to the MuseHub marketplace. muse domains info Inspect a single domain (agent-optimised). muse domains use Switch the active domain for this repo. muse domains validate Check protocol compliance. """ new: str | None = args.new as_json: bool = args.as_json if new is not None: _validate_domain_name(new) _scaffold_new_domain(new, as_json) return active_domain = _active_domain(_find_repo_root()) if as_json: _emit_json(active_domain) return _print_dashboard(active_domain)