"""muse config — local repository configuration. Provides structured, typed read/write access to ``.muse/config.toml``. For hub credentials and user identity, use ``muse auth``. For remote connections, use ``muse remote``. Settable namespaces -------------------- - ``hub.url`` — hub fabric URL (alias for ``muse hub connect``) - ``domain.*`` — domain-specific keys; read by the active plugin - ``limits.max_walk_commits`` — cap on merge-base BFS walk (int > 0) - ``limits.max_ancestors`` — cap on ancestor graph nodes (int > 0) - ``limits.max_graph_commits`` — cap on full graph traversal (int > 0) - ``limits.shard_prefix_length`` — object-store shard width: 2 or 4 Blocked via ``muse config set`` --------------------------------- - ``user.*`` — use ``muse auth register`` / ``muse auth whoami`` - ``auth.*`` — use ``muse auth register`` - ``remotes.*`` — use ``muse remote add/remove`` Security model -------------- - Domain key names are validated against TOML-structurally significant characters (``]``, ``[``, ``=``, ``"``, newlines) before being written, preventing TOML injection attacks via crafted key strings. - All user-controlled values are passed through ``sanitize_display()`` before being printed to prevent ANSI escape injection. - All diagnostic messages go to **stderr**; **stdout** is reserved for config values, JSON output, and TOML text. Output format ------------- ``muse config read`` emits TOML by default (human-readable). Pass ``--json`` for machine-readable output — no credentials are ever included. ``muse config get`` emits the raw value to stdout (one line, no quoting). With ``--json`` it emits ``{"key": "...", "value": "..."}``. ``muse config set`` confirms the write to stderr. With ``--json`` it emits ``{"status": "ok", "key": "...", "value": "...", "duration_ms": 0.001, "exit_code": 0}``. ``duration_ms`` Wall-clock time from argument parsing to output. ``exit_code`` Mirrors the process exit code (always ``0`` in the success paths). JSON schema for ``read``:: { "hub": {"url": "https://musehub.ai"}, "remotes": {"origin": "https://..."}, "domain": {"ticks_per_beat": "480"}, "limits": {"max_walk_commits": "10000"}, "duration_ms": 0.001, "exit_code": 0 } Examples -------- :: muse config read muse config read --json muse config get hub.url muse config get domain.ticks_per_beat --json muse config set hub.url https://musehub.ai muse config set domain.ticks_per_beat 480 muse config set limits.max_walk_commits 50000 muse config edit """ import argparse import json import logging import os import shlex import subprocess import sys import time from typing import TypedDict from muse.cli.config import ( ConfigTree, config_as_dict, config_path_for_editor, get_config_value, set_config_value, ) from muse.core.errors import ExitCode from muse.core.repo import find_repo_root from muse.core.validation import sanitize_display from muse.core.timing import start_timer from muse.core.envelope import EnvelopeJson, make_envelope logger = logging.getLogger(__name__) # ── TypedDicts ──────────────────────────────────────────────────────────────── class _ReadJson(EnvelopeJson): """JSON schema for ``muse config read --json`` (full config dump).""" config: ConfigTree class _GetJson(EnvelopeJson): """JSON schema for ``muse config get --json``.""" key: str value: str class _SetJson(EnvelopeJson): """JSON schema for ``muse config set --json``.""" status: str # "ok" key: str value: str # ── register ────────────────────────────────────────────────────────────────── def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``muse config`` subcommand tree. Every subcommand that produces structured output accepts ``--json``. All diagnostic messages go to stderr; stdout carries config data only. """ parser = subparsers.add_parser( "config", help="Local repository configuration.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) subs = parser.add_subparsers(dest="subcommand", metavar="SUBCOMMAND") subs.required = True edit_p = subs.add_parser( "edit", help="Open .muse/config.toml in $EDITOR or $VISUAL.", description=( "Open .muse/config.toml in your preferred editor.\n\n" "Editor resolution order:\n" " 1. $VISUAL (e.g. 'code --wait', 'vim')\n" " 2. $EDITOR (e.g. 'nano', 'emacs')\n" " 3. vi (hard fallback)\n\n" "Multi-word editor commands are supported:\n" " VISUAL='code --wait' muse config edit\n" " EDITOR='emacs -nw' muse config edit\n\n" "If .muse/config.toml does not exist it is created (empty) before\n" "the editor opens, so you can configure a fresh repository.\n\n" "Note: this command opens an interactive editor and is intended for\n" "human use. Agents should use 'muse config set' or 'muse config get'\n" "for programmatic config access.\n\n" "Exit codes: 0 success, 1 editor not found / editor exited non-zero,\n" " 2 not inside a Muse repository." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) edit_p.set_defaults(func=run_edit) get_p = subs.add_parser( "get", help="Print the value of a single config key.", description=( "Print the value of a single config key by dotted name.\n\n" "Exits non-zero when the key is not set — enabling agent branching:\n" " VALUE=$(muse config get hub.url) || echo 'not configured'\n\n" "Supported keys:\n" " hub.url\n" " domain. (any domain-specific key)\n" " limits.max_walk_commits, limits.max_ancestors,\n" " limits.max_graph_commits, limits.shard_prefix_length\n\n" "User identity (handle, type, email) is in identity.toml — use muse auth whoami.\n\n" "Agent quickstart:\n" " muse config get hub.url --json\n" " muse config get domain.ticks_per_beat --json | jq -r '.value'\n\n" "Exit codes: 0 key is set, 1 key not set or bad key format." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) get_p.add_argument( "key", metavar="KEY", help="Dotted key to read (e.g. hub.url, domain.ticks_per_beat, limits.max_walk_commits).", ) get_p.add_argument( "--json", "-j", action="store_true", dest="json_out", default=False, help="Emit {\"key\": \"...\", \"value\": \"...\"} JSON to stdout.", ) get_p.set_defaults(func=run_get) set_p = subs.add_parser( "set", help="Set a config value by dotted key.", description=( "Set a config value by dotted key (namespace.subkey).\n\n" "Settable namespaces:\n" " hub.url (HTTPS only)\n" " domain. (any domain-specific key)\n" " limits.max_walk_commits (int > 0)\n" " limits.max_ancestors (int > 0)\n" " limits.max_graph_commits (int > 0)\n" " limits.shard_prefix_length (2 or 4)\n\n" "Blocked namespaces — use the dedicated command instead:\n" " user.* → muse auth register / muse auth whoami\n" " auth.* → muse auth register\n" " remotes.* → muse remote add\n\n" "Agent quickstart:\n" " muse config set hub.url https://musehub.ai\n" " muse config set domain.ticks_per_beat 480\n" " muse config set limits.max_walk_commits 50000\n\n" "Exit codes: 0 success, 1 bad key / blocked namespace / validation error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) set_p.add_argument( "key", metavar="KEY", help="Dotted key to set (e.g. hub.url, domain.ticks_per_beat, limits.max_walk_commits).", ) set_p.add_argument("value", metavar="VALUE", help="New value.") set_p.add_argument( "--json", "-j", action="store_true", dest="json_out", default=False, help="Emit {\"status\": \"ok\", \"key\": \"...\", \"value\": \"...\"} JSON to stdout.", ) set_p.set_defaults(func=run_set) read_p = subs.add_parser( "read", help="Read the current repository configuration.", description=( "Read the full repository configuration from .muse/config.toml.\n\n" "Credentials are never included — the hub section only contains the URL.\n" "Text mode emits TOML to stdout; --json emits an indented JSON object.\n\n" "Agent quickstart:\n" " muse config show --json\n" " muse config show --json | jq '.hub.url'\n" " muse config show --json | jq '.limits.max_walk_commits'\n\n" "JSON schema:\n" " {\"hub\": {\"url\"},\n" " \"remotes\": {\"\": \"\"},\n" " \"domain\": {\"\": \"\"},\n" " \"limits\": {\"max_walk_commits\", \"max_ancestors\", ...}}\n\n" "Exit codes: 0 success (even when config is empty), 1 unknown --format." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) read_p.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit JSON instead of TOML text.", ) read_p.set_defaults(func=run_read) # ── run_read ────────────────────────────────────────────────────────────────── def run_read(args: argparse.Namespace) -> None: """Display the current repository configuration. Emits TOML by default. Credentials are never included. All user-controlled values are sanitised to prevent ANSI injection. JSON output goes to stdout; all diagnostics go to stderr. Agent quickstart ---------------- :: muse config read --json muse config get hub.url --json muse config set domain.ticks_per_beat 480 --json JSON fields ----------- hub Map with ``url`` key. remotes Map of remote name → URL. domain Domain-specific config map (e.g. ``ticks_per_beat``). limits Limits map (e.g. ``max_walk_commits``). Exit codes ---------- 0 Configuration displayed. 1 Invalid ``--format`` value. """ elapsed = start_timer() json_out: bool = args.json_out root = find_repo_root() data = config_as_dict(root) if json_out: print(json.dumps(_ReadJson(**make_envelope(elapsed), config=data))) return if not data: print("# No configuration set.") return hub = data.get("hub") if hub: print("[hub]") for key, val in sorted(hub.items()): print(f'{sanitize_display(key)} = "{sanitize_display(val)}"') print("") remotes = data.get("remotes") if remotes: for remote_name, remote_url in sorted(remotes.items()): print(f"[remotes.{sanitize_display(remote_name)}]") print(f'url = "{sanitize_display(remote_url)}"') print("") domain = data.get("domain") if domain: print("[domain]") for key, val in sorted(domain.items()): print(f'{sanitize_display(key)} = "{sanitize_display(val)}"') print("") limits = data.get("limits") if limits: print("[limits]") for key, val in sorted(limits.items()): print(f'{sanitize_display(key)} = {sanitize_display(val)}') print("") # ── run_get ─────────────────────────────────────────────────────────────────── def run_get(args: argparse.Namespace) -> None: """Print the value of a single config key. Key must be in ``namespace.subkey`` dotted form (e.g. ``hub.url``). Exits non-zero when the key is not set, enabling agent branching on config presence. Agent quickstart ---------------- :: muse config get hub.url --json muse config get domain.ticks_per_beat --json JSON fields ----------- key The key queried (e.g. ``"hub.url"``). value The string value stored. Exit codes ---------- 0 Key is set — value printed. 1 Key not set, bad format, or unknown namespace. """ elapsed = start_timer() key: str = args.key json_out: bool = args.json_out # ── Key format validation ───────────────────────────────────────────────── if "." not in key: print( f"❌ Key must be in 'namespace.subkey' form " f"(e.g. hub.url, domain.ticks_per_beat), got: {sanitize_display(key)!r}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) root = find_repo_root() value = get_config_value(key, root) if value is None: print(f"# {sanitize_display(key)} is not set", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if json_out: print(json.dumps(_GetJson(**make_envelope(elapsed), key=key, value=value))) return print(value) # ── run_set ─────────────────────────────────────────────────────────────────── def run_set(args: argparse.Namespace) -> None: """Set a config value by dotted key. Key must be in ``namespace.subkey`` dotted form. For credentials use ``muse auth register``; for remotes use ``muse remote add``. Diagnostics go to stderr; ``--json`` emits the result to stdout. Agent quickstart ---------------- :: muse config set hub.url https://musehub.ai --json muse config set domain.ticks_per_beat 480 --json muse config set limits.max_walk_commits 50000 --json JSON fields ----------- status ``"ok"`` on success. key The key written. value The value written. Exit codes ---------- 0 Value written successfully. 1 Bad key format, blocked namespace, or validation error. 2 Not inside a Muse repository. """ elapsed = start_timer() key: str = args.key value: str = args.value json_out: bool = args.json_out # ── Key format validation ───────────────────────────────────────────────── if "." not in key: print( f"❌ Key must be in 'namespace.subkey' form " f"(e.g. hub.url, domain.ticks_per_beat), got: {sanitize_display(key)!r}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) root = find_repo_root() try: set_config_value(key, value, root) except ValueError as exc: print(f"❌ {sanitize_display(str(exc))}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) from exc if json_out: print(json.dumps(_SetJson(**make_envelope(elapsed), status="ok", key=key, value=value))) else: print( f"✅ {sanitize_display(key)} = {sanitize_display(value)!r}", file=sys.stderr, ) # ── run_edit ────────────────────────────────────────────────────────────────── def run_edit(args: argparse.Namespace) -> None: """Open ``.muse/config.toml`` in ``$VISUAL``, ``$EDITOR``, or ``vi``. Editor resolution order: 1. ``$VISUAL`` — conventional for visual editors (e.g. ``code --wait``) 2. ``$EDITOR`` — conventional for terminal editors (e.g. ``nano``) 3. ``vi`` — hard fallback when neither variable is set Multi-word editor commands (e.g. ``VISUAL='code --wait'``) are parsed with :func:`shlex.split` and passed to :func:`subprocess.run` as a list so no shell is invoked — shell injection via a crafted ``$EDITOR`` value is not possible. If ``.muse/config.toml`` does not exist it is created empty before the editor opens. This lets users configure a fresh repository without first running ``muse config set``. All error messages go to stderr. Note: this command opens an interactive editor and is intended for human use. Agents should use ``muse config set`` / ``muse config get`` for programmatic config access. Exit codes: 0 Editor exited successfully. 1 Editor not found, or editor exited with a non-zero code. 2 Not inside a Muse repository. """ root = find_repo_root() if root is None: print("❌ Not inside a Muse repository.", file=sys.stderr) raise SystemExit(ExitCode.REPO_NOT_FOUND) config_path = config_path_for_editor(root) if not config_path.is_file(): # Create an empty config so the editor opens a valid (if empty) file. config_path.parent.mkdir(parents=True, exist_ok=True) config_path.write_text("") print( f"ℹ️ Created {sanitize_display(str(config_path))}", file=sys.stderr, ) raw_editor = os.environ.get("VISUAL") or os.environ.get("EDITOR") or "vi" try: editor_cmd = shlex.split(raw_editor) except ValueError as exc: print( f"❌ Could not parse editor command {sanitize_display(raw_editor)!r}: {exc}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) from exc try: subprocess.run(editor_cmd + [str(config_path)], check=True) except FileNotFoundError: print( f"❌ Editor not found: {sanitize_display(raw_editor)!r}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) except subprocess.CalledProcessError as exc: print(f"❌ Editor exited with code {exc.returncode}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR)