"""muse plumbing domain-info — inspect the active domain plugin. Reports which domain is active for this repository, which plugin class implements it, what optional capabilities it exposes, and the full structural schema it declares (merge mode, top-level element shape, dimensions). Output (JSON, default):: { "domain": "midi", "plugin_class": "MidiPlugin", "capabilities": { "structured_merge": true, "crdt": false, "rerere": false }, "schema": { "domain": "midi", "description": "...", "merge_mode": "three_way", "schema_version": "0.x.y", "top_level": { ... }, "dimensions": [ ... ] }, "registered_domains": ["bitcoin", "code", "midi", "scaffold"] } Text output (``--format text``):: Domain: midi Plugin: MidiPlugin Merge mode: three_way Capabilities: structured_merge Plumbing contract ----------------- - Exit 0: domain resolved and schema emitted. - Exit 1: no repository found; domain not registered; bad ``--format`` value. - Exit 3: plugin raised an unexpected error when computing its schema. Capabilities ------------ ``structured_merge`` The plugin implements ``StructuredMergePlugin`` and can perform symbol-level three-way merges rather than falling back to text-level line diffs. Agents should check this before attempting a semantic merge. ``crdt`` The plugin implements ``CRDTPlugin`` and exposes CRDT-annotated fields (e.g. ``reviewed_by`` ORSet, ``test_runs`` GCounter). Agents performing collaborative annotation should check this before writing CRDT fields. ``rerere`` The plugin implements ``RererePlugin`` and can record and replay conflict resolutions. Agents resolving conflicts in a swarm should check this to avoid duplicate resolution work. Agent use --------- Inspect any domain without entering its repository:: muse plumbing domain-info --domain midi muse plumbing domain-info --domain code --capabilities-only muse plumbing domain-info --all-domains """ from __future__ import annotations import argparse import json import logging import sys from typing import TypedDict import pathlib from muse.core.errors import ExitCode from muse.core.repo import require_repo from muse.core.validation import sanitize_display, validate_domain_name from muse.domain import CRDTPlugin, RererePlugin, StructuredMergePlugin from muse.plugins.registry import ( read_domain, registered_domains, resolve_plugin, resolve_plugin_by_domain, ) logger = logging.getLogger(__name__) _FORMAT_CHOICES = ("json", "text") class _CapabilitiesDict(TypedDict): structured_merge: bool crdt: bool rerere: bool def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the domain-info subcommand.""" parser = subparsers.add_parser( "domain-info", help="Inspect active domain plugin capabilities and schema.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--format", "-f", dest="fmt", default="json", metavar="FORMAT", help="Output format: json or text. (default: json)", ) parser.add_argument( "--json", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.", ) parser.add_argument( "--all-domains", "-a", action="store_true", dest="all_domains", help="List every registered domain instead of querying the active repo.", ) parser.add_argument( "--domain", "-d", default=None, dest="domain_name", metavar="DOMAIN", help=( "Inspect a specific domain by name without requiring an active repo. " "Example: --domain midi. " "Use --all-domains to list available names." ), ) parser.add_argument( "--capabilities-only", action="store_true", dest="capabilities_only", help=( "Emit only the capabilities dict. " "Ideal for agents doing merge-strategy negotiation — " "avoids serialising the full schema." ), ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Inspect the domain plugin active for this repository. Reports the domain name, plugin class, optional protocol capabilities (``StructuredMergePlugin``, ``CRDTPlugin``, ``RererePlugin``), and the full structural schema declared by the plugin. Use ``--all-domains`` to enumerate every domain registered in this Muse installation without requiring an active repository. Use ``--domain `` to inspect any registered domain by name without entering its repository — essential for agents doing capability negotiation across multiple domains in parallel. Use ``--capabilities-only`` to skip the full schema serialisation when all you need is the boolean capability flags. """ fmt: str = args.fmt all_domains: bool = args.all_domains domain_name: str | None = args.domain_name capabilities_only: bool = args.capabilities_only if fmt not in _FORMAT_CHOICES: print( json.dumps( {"error": f"Unknown format {fmt!r}. Valid: {', '.join(_FORMAT_CHOICES)}"} ), file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # --all-domains — no repo required. if all_domains: domains = registered_domains() if fmt == "text": for d in domains: print(d) else: print(json.dumps({"registered_domains": domains})) return # root is only available when resolving from a live repo (not --domain flag). root: pathlib.Path | None = None # --domain — inspect by name without entering a repo. if domain_name is not None: try: validate_domain_name(domain_name) except ValueError as exc: print(json.dumps({"error": f"Invalid domain name: {exc}"}), file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) try: plugin = resolve_plugin_by_domain(domain_name) except Exception as exc: print(json.dumps({"error": str(exc)}), file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) active_domain = domain_name else: root = require_repo() active_domain = read_domain(root) try: plugin = resolve_plugin(root) except Exception as exc: print(json.dumps({"error": str(exc)}), file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) plugin_class = type(plugin).__name__ capabilities: _CapabilitiesDict = { "structured_merge": isinstance(plugin, StructuredMergePlugin), "crdt": isinstance(plugin, CRDTPlugin), "rerere": isinstance(plugin, RererePlugin), } # --capabilities-only — skip schema serialisation. if capabilities_only: if fmt == "text": active_caps = [k for k, v in capabilities.items() if v] cap_str = ", ".join(active_caps) if active_caps else "none" print(f"Domain: {sanitize_display(active_domain)}") print(f"Capabilities: {cap_str}") else: print(json.dumps({ "domain": active_domain, "capabilities": dict(capabilities), })) return try: schema = plugin.schema() except Exception as exc: logger.debug("domain-info: plugin.schema() failed: %s", exc) print(json.dumps({"error": f"Plugin schema error: {exc}"}), file=sys.stderr) raise SystemExit(ExitCode.INTERNAL_ERROR) all_domains_list = registered_domains() # Knowtation vault stats — only when we have a live repo root. vault_stats: dict | None = None if active_domain == "knowtation" and root is not None: try: from muse.plugins.knowtation.stats import collect_vault_stats vault_stats = dict(collect_vault_stats(root)) except Exception as exc: logger.debug("domain-info: collect_vault_stats failed: %s", exc) vault_stats = {"error": str(exc)} if fmt == "text": print(f"Domain: {sanitize_display(active_domain)}") print(f"Plugin: {sanitize_display(plugin_class)}") print(f"Merge mode: {schema.get('merge_mode', 'unknown')}") active_caps = [k for k, v in capabilities.items() if v] cap_str = ", ".join(active_caps) if active_caps else "none" print(f"Capabilities: {cap_str}") print(f"Registered: {', '.join(all_domains_list)}") if vault_stats is not None: print(f"Notes: {vault_stats.get('note_count', '?')}") print(f"Projects: {', '.join(vault_stats.get('projects', [])) or '(none)'}") print(f"Tags: {vault_stats.get('tag_count', 0)}") print(f"Entities: {vault_stats.get('entity_count', 0)}") idx = vault_stats.get("index_freshness") print(f"Index: {idx if idx else 'not indexed'}") return out: dict = { "domain": active_domain, "plugin_class": plugin_class, "capabilities": dict(capabilities), "schema": dict(schema), "registered_domains": all_domains_list, } if vault_stats is not None: out["vault_stats"] = vault_stats print(json.dumps(out))