"""muse config — local repository configuration. Provides structured, typed read/write access to ``.muse/config.toml``. For hub credentials, use ``muse auth``. For remote connections, use ``muse remote``. Settable namespaces -------------------- - ``user.name`` — display name (human or agent handle) - ``user.email`` — contact email - ``user.type`` — ``"human"`` or ``"agent"`` - ``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`` --------------------------------- - ``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 show`` 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": "..."}``. JSON schema for ``show``:: { "user": {"name": "...", "email": "...", "type": "human|agent"}, "hub": {"url": "https://musehub.ai"}, "remotes": {"origin": "https://..."}, "domain": {"ticks_per_beat": "480"}, "limits": {"max_walk_commits": "10000"} } Examples -------- :: muse config show muse config show --json muse config get user.name muse config get user.type --json muse config set user.name "Alice" muse config set user.type agent --json muse config set domain.ticks_per_beat 480 muse config set limits.max_walk_commits 50000 muse config edit """ from __future__ import annotations import argparse import json import logging import os import shlex import subprocess import sys from typing import TypedDict from muse.cli.config import ( 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 logger = logging.getLogger(__name__) # ── TypedDicts ──────────────────────────────────────────────────────────────── class _GetJson(TypedDict): """JSON schema for ``muse config get --json``.""" key: str value: str class _SetJson(TypedDict): """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 user.type) || echo 'not configured'\n\n" "Supported keys:\n" " user.name, user.email, user.type\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" "Agent quickstart:\n" " muse config get hub.url --json\n" " muse config get user.type --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. user.name, hub.url, limits.max_walk_commits).", ) get_p.add_argument( "--json", "-j", action="store_true", dest="json_output", 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" " user.name, user.email, user.type\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" " auth.* → muse auth register\n" " remotes.* → muse remote add\n\n" "Agent quickstart:\n" " muse config set user.name Alice --json\n" " muse config set user.type agent\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. user.name, 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_output", default=False, help="Emit {\"status\": \"ok\", \"key\": \"...\", \"value\": \"...\"} JSON to stdout.", ) set_p.set_defaults(func=run_set) show_p = subs.add_parser( "show", help="Display the current repository configuration.", description=( "Display 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" " {\"user\": {\"name\", \"email\", \"type\"},\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, ) show_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON instead of TOML text.", ) show_p.add_argument( "--format", "-f", default="text", dest="fmt", help="Output format: text (default) or json (alias for --json).", ) show_p.set_defaults(func=run_show) # ── run_show ────────────────────────────────────────────────────────────────── def run_show(args: argparse.Namespace) -> None: """Display the current repository configuration. Emits TOML by default; use ``--json`` or ``--format json`` for agent-friendly structured output. Credentials are never included regardless of format. All user-controlled values (remote names, domain keys, hub URLs) are sanitized before being displayed in text mode to prevent ANSI injection. JSON output goes to stdout; all diagnostics go to stderr. JSON schema:: { "user": {"name": "...", "email": "...", "type": "human|agent"}, "hub": {"url": "https://musehub.ai"}, "remotes": {"origin": "https://..."}, "domain": {"ticks_per_beat": "480"}, "limits": {"max_walk_commits": "10000"} } """ json_output: bool = args.json_output fmt: str = args.fmt if fmt == "json": json_output = True elif fmt != "text": print( f"❌ Unknown --format '{sanitize_display(fmt)}'. Choose text or json.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) root = find_repo_root() data = config_as_dict(root) if json_output: print(json.dumps(data, indent=2)) return if not data: print("# No configuration set.") return user = data.get("user") if user: print("[user]") for key, val in sorted(user.items()): print(f'{sanitize_display(key)} = "{sanitize_display(val)}"') print("") 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. Exits non-zero when the key is not set or the key format is invalid, enabling agent branching:: VALUE=$(muse config get user.type) || echo "not set" muse config get user.name --json # {"key": "user.name", "value": "Alice"} muse config get hub.url # https://musehub.ai Key must be in ``namespace.subkey`` dotted form. A missing dot is a format error and gets a clear diagnostic rather than "not set". Exit codes: 0 Key is set — value printed to stdout. 1 Key not set, bad format, or unknown namespace. 2 Not inside a Muse repository (gracefully returns empty). """ key: str = args.key json_output: bool = args.json_output # ── Key format validation ───────────────────────────────────────────────── if "." not in key: print( f"❌ Key must be in 'namespace.subkey' form " f"(e.g. user.name, hub.url), 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_output: out: _GetJson = {"key": key, "value": value} print(json.dumps(out)) 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. A missing dot is a format error caught here before reaching ``set_config_value``, producing a clear diagnostic rather than a generic "unknown namespace" message. Domain keys are validated against TOML-structurally unsafe characters (``]``, ``[``, newlines, etc.) by ``set_config_value`` before being written. Diagnostics go to stderr; ``--json`` emits a result object to stdout:: muse config set user.name "Alice" muse config set user.type agent --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 For credentials, use ``muse auth register``. For remotes, use ``muse remote add``. Exit codes: 0 Value written successfully. 1 Bad key format, blocked namespace, or validation error. 2 Not inside a Muse repository. """ key: str = args.key value: str = args.value json_output: bool = args.json_output # ── Key format validation ───────────────────────────────────────────────── if "." not in key: print( f"❌ Key must be in 'namespace.subkey' form " f"(e.g. user.name, hub.url), 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_output: out: _SetJson = {"status": "ok", "key": key, "value": value} print(json.dumps(out)) 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)