aaronrene / gabriel-muse public
domain_info.py python
284 lines 9.5 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago
1 """muse plumbing domain-info — inspect the active domain plugin.
2
3 Reports which domain is active for this repository, which plugin class
4 implements it, what optional capabilities it exposes, and the full structural
5 schema it declares (merge mode, top-level element shape, dimensions).
6
7 Output (JSON, default)::
8
9 {
10 "domain": "midi",
11 "plugin_class": "MidiPlugin",
12 "capabilities": {
13 "structured_merge": true,
14 "crdt": false,
15 "rerere": false
16 },
17 "schema": {
18 "domain": "midi",
19 "description": "...",
20 "merge_mode": "three_way",
21 "schema_version": "0.x.y",
22 "top_level": { ... },
23 "dimensions": [ ... ]
24 },
25 "registered_domains": ["bitcoin", "code", "midi", "scaffold"]
26 }
27
28 Text output (``--format text``)::
29
30 Domain: midi
31 Plugin: MidiPlugin
32 Merge mode: three_way
33 Capabilities: structured_merge
34
35 Plumbing contract
36 -----------------
37
38 - Exit 0: domain resolved and schema emitted.
39 - Exit 1: no repository found; domain not registered; bad ``--format`` value.
40 - Exit 3: plugin raised an unexpected error when computing its schema.
41
42 Capabilities
43 ------------
44
45 ``structured_merge``
46 The plugin implements ``StructuredMergePlugin`` and can perform
47 symbol-level three-way merges rather than falling back to text-level
48 line diffs. Agents should check this before attempting a semantic merge.
49
50 ``crdt``
51 The plugin implements ``CRDTPlugin`` and exposes CRDT-annotated fields
52 (e.g. ``reviewed_by`` ORSet, ``test_runs`` GCounter). Agents performing
53 collaborative annotation should check this before writing CRDT fields.
54
55 ``rerere``
56 The plugin implements ``RererePlugin`` and can record and replay
57 conflict resolutions. Agents resolving conflicts in a swarm should
58 check this to avoid duplicate resolution work.
59
60 Agent use
61 ---------
62
63 Inspect any domain without entering its repository::
64
65 muse plumbing domain-info --domain midi
66 muse plumbing domain-info --domain code --capabilities-only
67 muse plumbing domain-info --all-domains
68 """
69
70 from __future__ import annotations
71
72 import argparse
73 import json
74 import logging
75 import sys
76 from typing import TypedDict
77
78 import pathlib
79
80 from muse.core.errors import ExitCode
81 from muse.core.repo import require_repo
82 from muse.core.validation import sanitize_display, validate_domain_name
83 from muse.domain import CRDTPlugin, RererePlugin, StructuredMergePlugin
84 from muse.plugins.registry import (
85 read_domain,
86 registered_domains,
87 resolve_plugin,
88 resolve_plugin_by_domain,
89 )
90
91 logger = logging.getLogger(__name__)
92
93 _FORMAT_CHOICES = ("json", "text")
94
95
96 class _CapabilitiesDict(TypedDict):
97 structured_merge: bool
98 crdt: bool
99 rerere: bool
100
101
102 def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
103 """Register the domain-info subcommand."""
104 parser = subparsers.add_parser(
105 "domain-info",
106 help="Inspect active domain plugin capabilities and schema.",
107 description=__doc__,
108 formatter_class=argparse.RawDescriptionHelpFormatter,
109 )
110 parser.add_argument(
111 "--format", "-f",
112 dest="fmt",
113 default="json",
114 metavar="FORMAT",
115 help="Output format: json or text. (default: json)",
116 )
117 parser.add_argument(
118 "--json", action="store_const", const="json", dest="fmt",
119 help="Shorthand for --format json.",
120 )
121 parser.add_argument(
122 "--all-domains", "-a",
123 action="store_true",
124 dest="all_domains",
125 help="List every registered domain instead of querying the active repo.",
126 )
127 parser.add_argument(
128 "--domain", "-d",
129 default=None,
130 dest="domain_name",
131 metavar="DOMAIN",
132 help=(
133 "Inspect a specific domain by name without requiring an active repo. "
134 "Example: --domain midi. "
135 "Use --all-domains to list available names."
136 ),
137 )
138 parser.add_argument(
139 "--capabilities-only",
140 action="store_true",
141 dest="capabilities_only",
142 help=(
143 "Emit only the capabilities dict. "
144 "Ideal for agents doing merge-strategy negotiation — "
145 "avoids serialising the full schema."
146 ),
147 )
148 parser.set_defaults(func=run)
149
150
151 def run(args: argparse.Namespace) -> None:
152 """Inspect the domain plugin active for this repository.
153
154 Reports the domain name, plugin class, optional protocol capabilities
155 (``StructuredMergePlugin``, ``CRDTPlugin``, ``RererePlugin``), and the
156 full structural schema declared by the plugin.
157
158 Use ``--all-domains`` to enumerate every domain registered in this Muse
159 installation without requiring an active repository.
160
161 Use ``--domain <name>`` to inspect any registered domain by name without
162 entering its repository — essential for agents doing capability negotiation
163 across multiple domains in parallel.
164
165 Use ``--capabilities-only`` to skip the full schema serialisation when all
166 you need is the boolean capability flags.
167 """
168 fmt: str = args.fmt
169 all_domains: bool = args.all_domains
170 domain_name: str | None = args.domain_name
171 capabilities_only: bool = args.capabilities_only
172
173 if fmt not in _FORMAT_CHOICES:
174 print(
175 json.dumps(
176 {"error": f"Unknown format {fmt!r}. Valid: {', '.join(_FORMAT_CHOICES)}"}
177 ),
178 file=sys.stderr,
179 )
180 raise SystemExit(ExitCode.USER_ERROR)
181
182 # --all-domains — no repo required.
183 if all_domains:
184 domains = registered_domains()
185 if fmt == "text":
186 for d in domains:
187 print(d)
188 else:
189 print(json.dumps({"registered_domains": domains}))
190 return
191
192 # root is only available when resolving from a live repo (not --domain flag).
193 root: pathlib.Path | None = None
194
195 # --domain <name> — inspect by name without entering a repo.
196 if domain_name is not None:
197 try:
198 validate_domain_name(domain_name)
199 except ValueError as exc:
200 print(json.dumps({"error": f"Invalid domain name: {exc}"}), file=sys.stderr)
201 raise SystemExit(ExitCode.USER_ERROR)
202 try:
203 plugin = resolve_plugin_by_domain(domain_name)
204 except Exception as exc:
205 print(json.dumps({"error": str(exc)}), file=sys.stderr)
206 raise SystemExit(ExitCode.USER_ERROR)
207 active_domain = domain_name
208 else:
209 root = require_repo()
210 active_domain = read_domain(root)
211 try:
212 plugin = resolve_plugin(root)
213 except Exception as exc:
214 print(json.dumps({"error": str(exc)}), file=sys.stderr)
215 raise SystemExit(ExitCode.USER_ERROR)
216
217 plugin_class = type(plugin).__name__
218
219 capabilities: _CapabilitiesDict = {
220 "structured_merge": isinstance(plugin, StructuredMergePlugin),
221 "crdt": isinstance(plugin, CRDTPlugin),
222 "rerere": isinstance(plugin, RererePlugin),
223 }
224
225 # --capabilities-only — skip schema serialisation.
226 if capabilities_only:
227 if fmt == "text":
228 active_caps = [k for k, v in capabilities.items() if v]
229 cap_str = ", ".join(active_caps) if active_caps else "none"
230 print(f"Domain: {sanitize_display(active_domain)}")
231 print(f"Capabilities: {cap_str}")
232 else:
233 print(json.dumps({
234 "domain": active_domain,
235 "capabilities": dict(capabilities),
236 }))
237 return
238
239 try:
240 schema = plugin.schema()
241 except Exception as exc:
242 logger.debug("domain-info: plugin.schema() failed: %s", exc)
243 print(json.dumps({"error": f"Plugin schema error: {exc}"}), file=sys.stderr)
244 raise SystemExit(ExitCode.INTERNAL_ERROR)
245
246 all_domains_list = registered_domains()
247
248 # Knowtation vault stats — only when we have a live repo root.
249 vault_stats: dict | None = None
250 if active_domain == "knowtation" and root is not None:
251 try:
252 from muse.plugins.knowtation.stats import collect_vault_stats
253 vault_stats = dict(collect_vault_stats(root))
254 except Exception as exc:
255 logger.debug("domain-info: collect_vault_stats failed: %s", exc)
256 vault_stats = {"error": str(exc)}
257
258 if fmt == "text":
259 print(f"Domain: {sanitize_display(active_domain)}")
260 print(f"Plugin: {sanitize_display(plugin_class)}")
261 print(f"Merge mode: {schema.get('merge_mode', 'unknown')}")
262 active_caps = [k for k, v in capabilities.items() if v]
263 cap_str = ", ".join(active_caps) if active_caps else "none"
264 print(f"Capabilities: {cap_str}")
265 print(f"Registered: {', '.join(all_domains_list)}")
266 if vault_stats is not None:
267 print(f"Notes: {vault_stats.get('note_count', '?')}")
268 print(f"Projects: {', '.join(vault_stats.get('projects', [])) or '(none)'}")
269 print(f"Tags: {vault_stats.get('tag_count', 0)}")
270 print(f"Entities: {vault_stats.get('entity_count', 0)}")
271 idx = vault_stats.get("index_freshness")
272 print(f"Index: {idx if idx else 'not indexed'}")
273 return
274
275 out: dict = {
276 "domain": active_domain,
277 "plugin_class": plugin_class,
278 "capabilities": dict(capabilities),
279 "schema": dict(schema),
280 "registered_domains": all_domains_list,
281 }
282 if vault_stats is not None:
283 out["vault_stats"] = vault_stats
284 print(json.dumps(out))
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago