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