"""Plugin registry — maps domain names to :class:`~muse.domain.MuseDomainPlugin` instances. Every CLI command that operates on domain state calls :func:`resolve_plugin` once to obtain the active plugin for the current repository. Adding support for a new domain requires only two changes: 1. Implement :class:`~muse.domain.MuseDomainPlugin` in a new module under ``muse/plugins//plugin.py``. 2. Register the plugin instance in ``_REGISTRY`` below. The domain for a repository is stored in ``.muse/repo.json`` under the key ``"domain"``. Repositories created before this key was introduced default to ``'midi'``. """ import os import pathlib from muse.core.types import load_json_file from muse.core.paths import repo_json_path as _repo_json_path from muse.core.errors import MuseCLIError from muse.core.schema import DomainSchema from muse.domain import MuseDomainPlugin from muse.plugins.code.plugin import CodePlugin from muse.plugins.identity.plugin import IdentityPlugin from muse.plugins.mist.plugin import MistPlugin from muse.plugins.scaffold.plugin import ScaffoldPlugin from muse.plugins.social.plugin import SocialPlugin from muse.plugins.timeline.plugin import TimelinePlugin from muse.plugins.todo.plugin import TodoPlugin type _PluginRegistry = dict[str, "MuseDomainPlugin"] _REGISTRY: _PluginRegistry = { "code": CodePlugin(), "identity": IdentityPlugin(), "mist": MistPlugin(), "social": SocialPlugin(), "scaffold": ScaffoldPlugin(), "timeline": TimelinePlugin(), "todo": TodoPlugin(), } # MIDI domain is suspended by default pending its own security and # performance audit — not sunset, just deferred while focus stays on the # code domain. The plugin itself is fully maintained; only its registration # is gated. Set MUSE_ENABLE_MIDI=1 to opt in locally (e.g. for a demo) — # never set in code shipped to end users. if os.environ.get("MUSE_ENABLE_MIDI"): from muse.plugins.midi.plugin import MidiPlugin _REGISTRY["midi"] = MidiPlugin() _DEFAULT_DOMAIN = "code" def _read_domain(root: pathlib.Path) -> str: """Return the domain name stored in ``.muse/repo.json``. Falls back to ``'midi'`` for repos that pre-date the ``domain`` field. """ data = load_json_file(_repo_json_path(root)) if data is None: return _DEFAULT_DOMAIN domain = data.get("domain") return str(domain) if domain else _DEFAULT_DOMAIN def resolve_plugin(root: pathlib.Path) -> MuseDomainPlugin: """Return the active domain plugin for the repository at *root*. Reads the ``"domain"`` key from ``.muse/repo.json`` and looks it up in the plugin registry. Raises :class:`~muse.core.errors.MuseCLIError` if the domain is not registered. Args: root: Repository root directory (contains ``.muse/``). Returns: The :class:`~muse.domain.MuseDomainPlugin` instance for this repo. Raises: MuseCLIError: When the domain stored in ``repo.json`` is not in the registry. This is a configuration error — either the plugin was not installed or ``repo.json`` was edited manually. """ domain = _read_domain(root) plugin = _REGISTRY.get(domain) if plugin is None: registered = ", ".join(sorted(_REGISTRY)) raise MuseCLIError( f"Unknown domain {domain!r}. Registered domains: {registered}" ) return plugin def read_domain(root: pathlib.Path) -> str: """Return the domain name for the repository at *root*. This is the same lookup used internally by :func:`resolve_plugin`. Use it when you need the domain string to construct a :class:`~muse.domain.SnapshotManifest` for a stored manifest. """ return _read_domain(root) def resolve_plugin_by_domain(domain: str) -> MuseDomainPlugin: """Return the plugin for *domain* without reading the filesystem. Use this when the caller has already read ``repo.json`` and only needs the plugin instance — avoids a redundant ``repo.json`` read compared to :func:`resolve_plugin`. Args: domain: Domain name string (e.g. ``'code'``). Returns: The :class:`~muse.domain.MuseDomainPlugin` instance for *domain*. Raises: MuseCLIError: When *domain* is not in the registry. """ plugin = _REGISTRY.get(domain) if plugin is None: registered = ", ".join(sorted(_REGISTRY)) raise MuseCLIError( f"Unknown domain {domain!r}. Registered domains: {registered}" ) return plugin def registered_domains() -> list[str]: """Return the sorted list of registered domain names.""" return sorted(_REGISTRY) def schema_for(domain: str) -> DomainSchema | None: """Return the ``DomainSchema`` for *domain*, or ``None`` if not registered. Allows the CLI and merge engine to look up a domain's schema without holding a plugin instance. Returns ``None`` rather than raising so callers can decide whether an unknown domain is an error or a soft miss. Args: domain: Domain name string (e.g. ``'midi'``). Returns: The :class:`~muse.core.schema.DomainSchema` declared by the plugin, or ``None`` if *domain* is not in the registry. """ plugin = _REGISTRY.get(domain) if plugin is None: return None return plugin.schema()