gabriel / muse public
registry.py python
142 lines 4.7 KB
Raw
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 73 days ago
1 """Plugin registry — maps domain names to :class:`~muse.domain.MuseDomainPlugin` instances.
2
3 Every CLI command that operates on domain state calls :func:`resolve_plugin`
4 once to obtain the active plugin for the current repository. Adding support
5 for a new domain requires only two changes:
6
7 1. Implement :class:`~muse.domain.MuseDomainPlugin` in a new module under
8 ``muse/plugins/<domain>/plugin.py``.
9 2. Register the plugin instance in ``_REGISTRY`` below.
10
11 The domain for a repository is stored in ``.muse/repo.json`` under the key
12 ``"domain"``. Repositories created before this key was introduced default to
13 ``'midi'``.
14 """
15
16 from __future__ import annotations
17
18 import json
19 import pathlib
20
21 from muse.core.errors import MuseCLIError
22 from muse.core.schema import DomainSchema
23 from muse.domain import MuseDomainPlugin
24 from muse.plugins.code.plugin import CodePlugin
25 from muse.plugins.scaffold.plugin import ScaffoldPlugin
26
27
28 type _PluginRegistry = dict[str, "MuseDomainPlugin"]
29 # MIDI domain is temporarily suspended pending its own security and
30 # performance audit. Re-enable by restoring the MidiPlugin import and
31 # the "midi" entry in _REGISTRY below.
32 # from muse.plugins.midi.plugin import MidiPlugin
33
34 _REGISTRY: _PluginRegistry = {
35 "code": CodePlugin(),
36 # "midi": MidiPlugin(), # suspended — see comment above
37 "scaffold": ScaffoldPlugin(),
38 }
39
40 _DEFAULT_DOMAIN = "code"
41
42
43 def _read_domain(root: pathlib.Path) -> str:
44 """Return the domain name stored in ``.muse/repo.json``.
45
46 Falls back to ``'midi'`` for repos that pre-date the ``domain`` field.
47 """
48 repo_json = root / ".muse" / "repo.json"
49 try:
50 data = json.loads(repo_json.read_text(encoding="utf-8"))
51 domain = data.get("domain")
52 return str(domain) if domain else _DEFAULT_DOMAIN
53 except (OSError, json.JSONDecodeError):
54 return _DEFAULT_DOMAIN
55
56
57 def resolve_plugin(root: pathlib.Path) -> MuseDomainPlugin:
58 """Return the active domain plugin for the repository at *root*.
59
60 Reads the ``"domain"`` key from ``.muse/repo.json`` and looks it up in
61 the plugin registry. Raises :class:`~muse.core.errors.MuseCLIError` if
62 the domain is not registered.
63
64 Args:
65 root: Repository root directory (contains ``.muse/``).
66
67 Returns:
68 The :class:`~muse.domain.MuseDomainPlugin` instance for this repo.
69
70 Raises:
71 MuseCLIError: When the domain stored in ``repo.json`` is not in the
72 registry. This is a configuration error — either the plugin was
73 not installed or ``repo.json`` was edited manually.
74 """
75 domain = _read_domain(root)
76 plugin = _REGISTRY.get(domain)
77 if plugin is None:
78 registered = ", ".join(sorted(_REGISTRY))
79 raise MuseCLIError(
80 f"Unknown domain {domain!r}. Registered domains: {registered}"
81 )
82 return plugin
83
84
85 def read_domain(root: pathlib.Path) -> str:
86 """Return the domain name for the repository at *root*.
87
88 This is the same lookup used internally by :func:`resolve_plugin`.
89 Use it when you need the domain string to construct a
90 :class:`~muse.domain.SnapshotManifest` for a stored manifest.
91 """
92 return _read_domain(root)
93
94
95 def resolve_plugin_by_domain(domain: str) -> MuseDomainPlugin:
96 """Return the plugin for *domain* without reading the filesystem.
97
98 Use this when the caller has already read ``repo.json`` and only needs
99 the plugin instance — avoids a redundant ``repo.json`` read compared to
100 :func:`resolve_plugin`.
101
102 Args:
103 domain: Domain name string (e.g. ``'code'``).
104
105 Returns:
106 The :class:`~muse.domain.MuseDomainPlugin` instance for *domain*.
107
108 Raises:
109 MuseCLIError: When *domain* is not in the registry.
110 """
111 plugin = _REGISTRY.get(domain)
112 if plugin is None:
113 registered = ", ".join(sorted(_REGISTRY))
114 raise MuseCLIError(
115 f"Unknown domain {domain!r}. Registered domains: {registered}"
116 )
117 return plugin
118
119
120 def registered_domains() -> list[str]:
121 """Return the sorted list of registered domain names."""
122 return sorted(_REGISTRY)
123
124
125 def schema_for(domain: str) -> DomainSchema | None:
126 """Return the ``DomainSchema`` for *domain*, or ``None`` if not registered.
127
128 Allows the CLI and merge engine to look up a domain's schema without
129 holding a plugin instance. Returns ``None`` rather than raising so callers
130 can decide whether an unknown domain is an error or a soft miss.
131
132 Args:
133 domain: Domain name string (e.g. ``'midi'``).
134
135 Returns:
136 The :class:`~muse.core.schema.DomainSchema` declared by the plugin,
137 or ``None`` if *domain* is not in the registry.
138 """
139 plugin = _REGISTRY.get(domain)
140 if plugin is None:
141 return None
142 return plugin.schema()
File History 1 commit
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 73 days ago