"""Muse CLI configuration helpers. Reads and writes ``.muse/config.toml`` — the per-repository configuration file. Credentials and user identity are **not** stored here; they live in ``~/.muse/identity.toml`` managed by :mod:`muse.core.identity`. Config schema ------------- :: [hub] url = "https://musehub.ai" # MuseHub fabric endpoint for this repo [remotes.origin] url = "https://hub.muse.io/repos/my-repo" branch = "main" [domain] # Domain-specific key/value pairs; read by the active domain plugin. # ticks_per_beat = "480" Settable via ``muse config set`` --------------------------------- - ``hub.url`` (alias: ``muse hub connect ``) - ``domain.*`` Not settable via ``muse config set`` -------------------------------------- - ``user.*`` — use ``muse auth register`` / ``muse auth whoami`` - ``remotes.*`` — use ``muse remote add/remove`` - credentials — use ``muse auth register`` Token resolution ---------------- :func:`get_signing_identity` reads the hub URL from this file, then resolves the signing identity from ``~/.muse/identity.toml`` via :func:`muse.core.identity.resolve_token`. The token is **never** logged. """ import fnmatch import logging import pathlib import shutil import subprocess import tomllib from typing import TypedDict from muse.core.types import short_id from muse.core.paths import config_toml_path as _config_toml_path, user_muse_dir as _user_muse_dir, user_config_toml_path as _user_config_toml_path, remote_tracking_dir as _remote_tracking_dir, remote_ref_path as _remote_ref_path from muse.core.refs import read_ref from muse.core.io import write_text_atomic logger = logging.getLogger(__name__) type RemotesMap = dict[str, RemoteEntry] # remote_name → remote entry type DomainConfig = dict[str, str] # domain key → value type ConfigSection = dict[str, str] # generic flattened section dict type ConfigTree = dict[str, ConfigSection] # section → key → value (for JSON output) type DefaultsMap = dict[str, int] # config key → default int value type _SecurityConfig = dict[str, list[str]] # security section from global config # --------------------------------------------------------------------------- # Named configuration types # --------------------------------------------------------------------------- class HubConfig(TypedDict, total=False): """``[hub]`` section in ``.muse/config.toml``.""" url: str class RemoteEntry(TypedDict, total=False): """``[remotes.]`` section in ``.muse/config.toml``.""" url: str branch: str promisor: bool # when False, this remote is not used as a promisor for missing objects class LimitsConfig(TypedDict, total=False): """``[limits]`` section in ``.muse/config.toml``. All values are optional — defaults are used when absent. Keys map to the ``[limits]`` TOML table:: [limits] max_walk_commits = 10000 # cap for walk_commits_between / muse log max_ancestors = 50000 # cap for find_merge_base BFS max_graph_commits = 50000 # cap for _collect_all_commits (--graph --all) shard_prefix_length = 2 # object store shard depth: 2 (256 shards) # or 4 (65536 shards) for very large repos """ max_walk_commits: int max_ancestors: int max_graph_commits: int shard_prefix_length: int class CommitConfig(TypedDict, total=False): """``[commit]`` section in ``.muse/config.toml``.""" sign: bool class SymlogConfig(TypedDict, total=False): """``[symlog]`` section in ``.muse/config.toml``. Controls default TTL for per-symbol journal expiry:: [symlog] expire_days = 90 # entries older than this are pruned by muse gc / muse symlog expire """ expire_days: int class PushConfig(TypedDict, total=False): """``[push]`` section in ``.muse/config.toml``.""" tags: bool # when true, version tags are pushed on every muse push reflog: bool # when true, this branch's reflog is pushed on every muse push (muse#54 item 4) symlog: bool # when true, every local symbol journal is pushed on every muse push (muse#53 item 4) class FetchConfig(TypedDict, total=False): """``[fetch]`` section in ``.muse/config.toml``. Mirrors git's ``fetch.prune`` exactly (muse#22):: [fetch] prune = true # muse fetch / muse pull auto-prune stale remote-tracking refs """ prune: bool class BranchMeta(TypedDict, total=False): """Per-branch metadata stored under ``[branch.""]`` in config.toml. Fields written by ``muse branch --intent / --resumable``:: [branch."feat/my-thing"] intent = "refactor auth layer" resumable = true Fields written by ``muse push`` upstream tracking (preserved on read/write):: remote = "origin" merge = "refs/heads/feat/my-thing" Fields written once, at branch-creation time, by ``muse switch -c`` / ``muse checkout -b`` / ``muse branch `` — never touched again afterward (musehub#216: this is a creation record, distinct from whoever authored the branch's current tip commit):: created_by_handle = "gabriel" created_by_kind = "human" # "human" | "agent" created_at = "2026-10-22T19:00:00+00:00" """ intent: str # short description of what this branch is for resumable: bool # true when this branch is a resumable agent checkpoint remote: str # upstream remote name (e.g. "origin") merge: str # upstream merge ref (e.g. "refs/heads/main") created_by_handle: str # who created this branch (musehub#216) created_by_kind: str # "human" | "agent" created_at: str # ISO 8601 creation timestamp class MuseConfig(TypedDict, total=False): """Structured view of the entire ``.muse/config.toml`` file.""" hub: HubConfig remotes: RemotesMap symlog: SymlogConfig domain: DomainConfig limits: LimitsConfig commit: CommitConfig push: PushConfig fetch: FetchConfig branch: "dict[str, BranchMeta]" # branch_name → per-branch metadata protected_branches: "list[str]" # fnmatch patterns from [protected_branches] class RemoteConfig(TypedDict, total=False): """Public-facing remote descriptor returned by :func:`list_remotes`.""" name: str # always present url: str # always present # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _config_path(repo_root: pathlib.Path | None) -> pathlib.Path: root = (repo_root or pathlib.Path.cwd()).resolve() return _config_toml_path(root) def _load_config(config_path: pathlib.Path) -> MuseConfig: """Load and parse config.toml; return an empty MuseConfig if absent.""" if not config_path.is_file(): return {} try: with config_path.open("rb") as fh: raw = tomllib.load(fh) except Exception as exc: # noqa: BLE001 logger.warning("⚠️ Failed to parse %s: %s", config_path, exc) return {} config: MuseConfig = {} hub_raw = raw.get("hub") if isinstance(hub_raw, dict): hub: HubConfig = {} url_val = hub_raw.get("url") if isinstance(url_val, str): hub["url"] = url_val config["hub"] = hub remotes_raw = raw.get("remotes") if isinstance(remotes_raw, dict): remotes: RemotesMap = {} for name, remote_raw in remotes_raw.items(): if isinstance(remote_raw, dict): entry: RemoteEntry = {} rurl = remote_raw.get("url") if isinstance(rurl, str): entry["url"] = rurl branch_val = remote_raw.get("branch") if isinstance(branch_val, str): entry["branch"] = branch_val promisor_val = remote_raw.get("promisor") if isinstance(promisor_val, bool): entry["promisor"] = promisor_val remotes[name] = entry config["remotes"] = remotes domain_raw = raw.get("domain") if isinstance(domain_raw, dict): domain: DomainConfig = {} for key, val in domain_raw.items(): if isinstance(val, str): domain[key] = val config["domain"] = domain limits_raw = raw.get("limits") if isinstance(limits_raw, dict): limits: LimitsConfig = {} mwc = limits_raw.get("max_walk_commits") if isinstance(mwc, int) and mwc > 0: limits["max_walk_commits"] = mwc ma = limits_raw.get("max_ancestors") if isinstance(ma, int) and ma > 0: limits["max_ancestors"] = ma mgc = limits_raw.get("max_graph_commits") if isinstance(mgc, int) and mgc > 0: limits["max_graph_commits"] = mgc spl = limits_raw.get("shard_prefix_length") if isinstance(spl, int) and spl in (2, 4): limits["shard_prefix_length"] = spl config["limits"] = limits commit_raw = raw.get("commit") if isinstance(commit_raw, dict): commit_cfg: CommitConfig = {} sign_v = commit_raw.get("sign") if isinstance(sign_v, bool): commit_cfg["sign"] = sign_v if commit_cfg: config["commit"] = commit_cfg reflog_raw = raw.get("reflog") if isinstance(reflog_raw, dict): reflog_cfg: dict = {} ed = reflog_raw.get("expire_days") if isinstance(ed, int) and ed > 0: reflog_cfg["expire_days"] = ed if reflog_cfg: config["reflog"] = reflog_cfg symlog_raw = raw.get("symlog") if isinstance(symlog_raw, dict): symlog_cfg: SymlogConfig = {} sl_ed = symlog_raw.get("expire_days") if isinstance(sl_ed, int) and sl_ed > 0: symlog_cfg["expire_days"] = sl_ed if symlog_cfg: config["symlog"] = symlog_cfg push_raw = raw.get("push") if isinstance(push_raw, dict): push_cfg: PushConfig = {} tags_v = push_raw.get("tags") if isinstance(tags_v, bool): push_cfg["tags"] = tags_v reflog_v = push_raw.get("reflog") if isinstance(reflog_v, bool): push_cfg["reflog"] = reflog_v symlog_v = push_raw.get("symlog") if isinstance(symlog_v, bool): push_cfg["symlog"] = symlog_v if push_cfg: config["push"] = push_cfg fetch_raw = raw.get("fetch") if isinstance(fetch_raw, dict): fetch_cfg: FetchConfig = {} prune_v = fetch_raw.get("prune") if isinstance(prune_v, bool): fetch_cfg["prune"] = prune_v if fetch_cfg: config["fetch"] = fetch_cfg branch_raw = raw.get("branch") if isinstance(branch_raw, dict): branch_map: dict[str, BranchMeta] = {} for bname, bdata in branch_raw.items(): if not isinstance(bdata, dict): continue bmeta: BranchMeta = {} intent_v = bdata.get("intent") if isinstance(intent_v, str): bmeta["intent"] = intent_v resumable_v = bdata.get("resumable") if isinstance(resumable_v, bool): bmeta["resumable"] = resumable_v remote_v = bdata.get("remote") if isinstance(remote_v, str): bmeta["remote"] = remote_v merge_v = bdata.get("merge") if isinstance(merge_v, str): bmeta["merge"] = merge_v created_by_handle_v = bdata.get("created_by_handle") if isinstance(created_by_handle_v, str): bmeta["created_by_handle"] = created_by_handle_v created_by_kind_v = bdata.get("created_by_kind") if isinstance(created_by_kind_v, str): bmeta["created_by_kind"] = created_by_kind_v created_at_v = bdata.get("created_at") if isinstance(created_at_v, str): bmeta["created_at"] = created_at_v branch_map[bname] = bmeta if branch_map: config["branch"] = branch_map pb_raw = raw.get("protected_branches") if isinstance(pb_raw, dict): branches_val = pb_raw.get("branches") if isinstance(branches_val, list): patterns = [p for p in branches_val if isinstance(p, str)] config["protected_branches"] = patterns return config def _escape(value: str) -> str: """Escape a TOML basic string value (backslash and double-quote only). TOML basic strings allow control characters escaped as ``\\n``, ``\\t``, etc., but we store only printable content — control characters in values are also stripped here so the resulting TOML file remains parseable. """ return ( value.replace("\\", "\\\\") .replace('"', '\\"') .replace("\n", "\\n") .replace("\r", "\\r") .replace("\0", "") ) # Characters that are structurally significant in unquoted TOML keys and # table headers. Any of these in a key name would allow injection of # arbitrary TOML sections or key-value pairs. _TOML_KEY_UNSAFE: frozenset[str] = frozenset('\n\r\0][="') def _validate_toml_key(key: str, context: str = "key") -> None: """Raise ``ValueError`` if *key* contains TOML-structurally unsafe characters. Prevents injection attacks where a crafted key like ``x]\\n[injected`` would break the TOML section structure and allow writing arbitrary sections. Args: key: Key string to validate. context: Human-readable label used in the error message (e.g. ``"domain key"``). Raises: ValueError: If any character in *key* is in ``_TOML_KEY_UNSAFE``. """ bad = _TOML_KEY_UNSAFE & set(key) if bad: chars = ", ".join(sorted(repr(c) for c in bad)) raise ValueError( f"Config {context} {key!r} contains characters not allowed in TOML keys: {chars}" ) def _dump_toml(config: MuseConfig) -> str: """Serialise a MuseConfig to TOML text. Section order: ``[hub]``, ``[remotes.*]``, ``[domain]``, ``[limits]``. All key names are validated against ``_TOML_KEY_UNSAFE`` before being written, preventing TOML injection via crafted domain keys or remote names. """ lines: list[str] = [] hub = config.get("hub") if hub: lines.append("[hub]") url = hub.get("url", "") if url: lines.append(f'url = "{_escape(url)}"') lines.append("") remotes = config.get("remotes") or {} for remote_name in sorted(remotes): # Remote names come from _load_config which parses TOML, so they are # safe at read time. Validate defensively before writing. _validate_toml_key(remote_name, "remote name") entry = remotes[remote_name] lines.append(f"[remotes.{remote_name}]") rurl = entry.get("url", "") if rurl: lines.append(f'url = "{_escape(rurl)}"') branch = entry.get("branch", "") if branch: lines.append(f'branch = "{_escape(branch)}"') if "promisor" in entry: lines.append(f'promisor = {"true" if entry["promisor"] else "false"}') lines.append("") domain = config.get("domain") or {} if domain: lines.append("[domain]") for key, val in sorted(domain.items()): _validate_toml_key(key, "domain key") lines.append(f'{key} = "{_escape(val)}"') lines.append("") limits = config.get("limits") or {} if limits: lines.append("[limits]") mwc = limits.get("max_walk_commits") if mwc is not None: lines.append(f"max_walk_commits = {mwc}") ma = limits.get("max_ancestors") if ma is not None: lines.append(f"max_ancestors = {ma}") mgc = limits.get("max_graph_commits") if mgc is not None: lines.append(f"max_graph_commits = {mgc}") spl = limits.get("shard_prefix_length") if spl is not None: lines.append(f"shard_prefix_length = {spl}") lines.append("") commit_cfg = config.get("commit") or {} if commit_cfg: lines.append("[commit]") if "sign" in commit_cfg: lines.append(f"sign = {'true' if commit_cfg['sign'] else 'false'}") lines.append("") reflog_cfg = config.get("reflog") or {} if reflog_cfg: lines.append("[reflog]") ed = reflog_cfg.get("expire_days") if ed is not None: lines.append(f"expire_days = {ed}") lines.append("") symlog_cfg_out = config.get("symlog") or {} if symlog_cfg_out: lines.append("[symlog]") sl_ed = symlog_cfg_out.get("expire_days") if sl_ed is not None: lines.append(f"expire_days = {sl_ed}") lines.append("") push_cfg_out = config.get("push") or {} if push_cfg_out: lines.append("[push]") if "tags" in push_cfg_out: lines.append(f"tags = {'true' if push_cfg_out['tags'] else 'false'}") if "reflog" in push_cfg_out: lines.append(f"reflog = {'true' if push_cfg_out['reflog'] else 'false'}") if "symlog" in push_cfg_out: lines.append(f"symlog = {'true' if push_cfg_out['symlog'] else 'false'}") lines.append("") fetch_cfg_out = config.get("fetch") or {} if fetch_cfg_out: lines.append("[fetch]") if "prune" in fetch_cfg_out: lines.append(f"prune = {'true' if fetch_cfg_out['prune'] else 'false'}") lines.append("") branch_sections = config.get("branch") or {} for bname in sorted(branch_sections): _validate_toml_key(bname, "branch name") bmeta = branch_sections[bname] # Skip empty metadata dicts — no section needed. if not bmeta: continue # Branch names require quoted keys (may contain '/' and other chars # that are not valid in bare TOML keys). lines.append(f'[branch."{_escape(bname)}"]') intent = bmeta.get("intent", "") if intent: lines.append(f'intent = "{_escape(intent)}"') if "resumable" in bmeta: lines.append(f"resumable = {'true' if bmeta['resumable'] else 'false'}") remote = bmeta.get("remote", "") if remote: lines.append(f'remote = "{_escape(remote)}"') merge = bmeta.get("merge", "") if merge: lines.append(f'merge = "{_escape(merge)}"') created_by_handle = bmeta.get("created_by_handle", "") if created_by_handle: lines.append(f'created_by_handle = "{_escape(created_by_handle)}"') created_by_kind = bmeta.get("created_by_kind", "") if created_by_kind: lines.append(f'created_by_kind = "{_escape(created_by_kind)}"') created_at = bmeta.get("created_at", "") if created_at: lines.append(f'created_at = "{_escape(created_at)}"') lines.append("") return "\n".join(lines) # --------------------------------------------------------------------------- # Auth token resolution (via identity store) # --------------------------------------------------------------------------- def get_signing_identity( repo_root: pathlib.Path | None = None, remote_url: str | None = None, agent_id: str | None = None, ) -> "object | None": """Return a :class:`~muse.core.transport.SigningIdentity` for a hub, or ``None``. Resolution order: 1. ``MUSE_AGENT_KEY_FD`` environment variable — integer file descriptor from which exactly 64 bytes of sub-seed are read (then the fd is closed). The Ed25519 identity key is derived via :func:`~muse.core.hdkeys.derive_identity_key`, scoped to whichever hub is resolved from *remote_url* / ``[hub] url`` (musehub#221) — if no hub is resolvable, this branch is skipped entirely rather than deriving a hub-independent key. The handle is taken from ``MUSE_AGENT_HANDLE`` (defaults to *agent_id* if set, else ``"agent"``). This is the only supported env-based injection mechanism; the secret travels through the kernel pipe buffer and never appears in ``/proc//environ``. 2. Agent-specific entry in ``~/.muse/identity.toml`` keyed by ``"hostname#agent_id"`` — when *agent_id* is provided. 3. Human entry in ``~/.muse/identity.toml`` keyed by bare hostname. 4. Hub URL from ``[hub] url`` in ``.muse/config.toml`` (fallback lookup URL when *remote_url* is not supplied). The private key is **never** logged. Args: repo_root: Repository root. Defaults to ``Path.cwd()``. remote_url: URL of the specific remote being contacted. agent_id: Agent handle, e.g. ``"agentception-abc123"``. Used to try an agent-specific key before falling back to the human key. Returns: :class:`~muse.core.transport.SigningIdentity` or ``None``. """ import os as _os from muse.core.identity import resolve_signing_identity, hostname_from_url # avoid circular import from muse.core.transport import SigningIdentity from muse.core.hdkeys import hub_index # Resolved once, up front — every derivation below is scoped to this hub # (musehub#221). Without a resolvable hub, MUSE_AGENT_KEY_FD cannot # safely derive a key: there is no legitimate hub-independent identity # key, so we fail closed and fall through to the identity-store lookup # (which will itself return None below if there's truly no hub context). lookup_url: str | None = remote_url or get_hub_url(repo_root) # 1. MUSE_AGENT_KEY_FD — read 64-byte sub-seed from a pipe fd. # This is the only supported env-var injection mechanism. # The secret travels through the kernel pipe buffer and never appears # in /proc//environ. key_fd_str = _os.environ.get("MUSE_AGENT_KEY_FD", "").strip() if key_fd_str and lookup_url is not None: try: key_fd = int(key_fd_str) import os as _os2 sub_seed = bytearray(_os2.read(key_fd, 64)) _os2.close(key_fd) if len(sub_seed) == 64: from muse.core.hdkeys import derive_identity_key hub = hub_index(hostname_from_url(lookup_url)) dk = derive_identity_key(sub_seed, hub=hub) from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey private_key = Ed25519PrivateKey.from_private_bytes(dk.private_bytes) dk.zero() sub_seed[:] = b"\x00" * len(sub_seed) handle = ( _os.environ.get("MUSE_AGENT_HANDLE", "").strip() or agent_id or "agent" ) logger.debug("✅ Signing identity from MUSE_AGENT_KEY_FD (handle=%s)", handle) return SigningIdentity(handle=handle, private_key=private_key) logger.warning( "⚠️ MUSE_AGENT_KEY_FD fd=%s yielded %d bytes (expected 64) — falling through", key_fd_str, len(sub_seed), ) except Exception as exc: logger.warning("⚠️ MUSE_AGENT_KEY_FD could not be read: %s — falling through", exc) elif key_fd_str: logger.warning( "⚠️ MUSE_AGENT_KEY_FD set but no hub is resolvable (no remote_url, no " "[hub] url) — cannot derive a hub-scoped key, falling through" ) # 2. Identity store lookup (agent key → human key fallback). if lookup_url is None: logger.debug("⚠️ No hub configured — skipping signing identity lookup") return None result = resolve_signing_identity(lookup_url, agent_id=agent_id) if result is None: logger.debug( "⚠️ No signing identity for hub %s — run `muse auth keygen && muse auth register`", lookup_url, ) return None handle, private_key = result logger.debug("✅ Signing identity resolved for hub %s (handle=%s)", lookup_url, handle) return SigningIdentity(handle=handle, private_key=private_key) # --------------------------------------------------------------------------- # Hub helpers # --------------------------------------------------------------------------- def get_hub_url(repo_root: pathlib.Path | None = None) -> str | None: """Return the hub URL from ``[hub] url``, or ``None`` if not configured. Resolution order: 1. ``/.muse/config.toml`` — repo-local config (highest priority). 2. ``~/.muse/config.toml`` — global user config (fallback). This fallback ensures ``muse auth whoami`` and other hub-aware commands work without ``--hub`` even when invoked outside a repository, as long as the user has set a default hub in their global config. Args: repo_root: Repository root. Defaults to ``Path.cwd()``. Returns: URL string, or ``None``. """ config = _load_config(_config_path(repo_root)) hub = config.get("hub") if hub is not None: url = hub.get("url", "") if url.strip(): return url.strip() # Fall back to ~/.muse/config.toml so hub-aware commands (e.g. `muse auth # whoami`) work without --hub when invoked outside a repository. global_config = _load_config(_GLOBAL_CONFIG_FILE) global_hub = global_config.get("hub") if global_hub is not None: url = global_hub.get("url", "") if url.strip(): return url.strip() return None def set_hub_url(url: str, repo_root: pathlib.Path | None = None) -> None: """Write ``[hub] url`` to ``.muse/config.toml``. Preserves all other sections. Creates the config file if absent. Rejects ``http://`` URLs — Muse never contacts a hub over cleartext HTTP. Args: url: Hub URL (must be ``https://``). repo_root: Repository root. Defaults to ``Path.cwd()``. Raises: ValueError: If *url* does not use the ``https://`` scheme. """ _is_loopback = url.startswith("http://localhost") or url.startswith("http://127.0.0.1") or url.startswith("http://[::1]") if not url.startswith("https://") and not _is_loopback: raise ValueError( f"Hub URL must use HTTPS. Got: {url!r}\n" "Muse never connects to a hub over cleartext HTTP.\n" "(Exception: http://localhost and http://127.0.0.1 are allowed for local development.)" ) cp = _config_path(repo_root) cp.parent.mkdir(parents=True, exist_ok=True) config = _load_config(cp) config["hub"] = HubConfig(url=url) write_text_atomic(cp, _dump_toml(config)) logger.info("✅ Hub URL set to %s", url) def clear_hub_url(repo_root: pathlib.Path | None = None) -> None: """Remove the ``[hub]`` section from ``.muse/config.toml``. Args: repo_root: Repository root. Defaults to ``Path.cwd()``. """ cp = _config_path(repo_root) config = _load_config(cp) if "hub" in config: del config["hub"] write_text_atomic(cp, _dump_toml(config)) logger.info("✅ Hub disconnected") # --------------------------------------------------------------------------- # Generic dotted-key helpers # --------------------------------------------------------------------------- _BlockedNS = dict[str, str] _BLOCKED_NAMESPACES: _BlockedNS = { "auth": "Use `muse auth keygen` and `muse auth register` to manage credentials.", "remotes": "Use `muse remote add/remove/rename` to manage remotes.", "user": "User identity is managed via `muse auth register`. Run `muse auth whoami` to inspect.", } _SETTABLE_NAMESPACES = {"hub", "domain", "limits", "commit", "reflog", "symlog", "push", "fetch"} # Default cap values — used when the [limits] section is absent or the key # is not set. These are the same values that were previously hardcoded inside # the individual functions. _DEFAULT_MAX_WALK_COMMITS: int = 10_000 _DEFAULT_MAX_ANCESTORS: int = 50_000 _DEFAULT_MAX_GRAPH_COMMITS: int = 50_000 _DEFAULT_SHARD_PREFIX_LENGTH: int = 2 def get_limit(key: str, repo_root: pathlib.Path | None = None) -> int: """Return a ``[limits]`` integer cap from config, or its default. Args: key: Limit key — one of ``max_walk_commits``, ``max_ancestors``, ``max_graph_commits``. repo_root: Repository root; ``None`` falls back to ``Path.cwd()``. Returns: Configured integer value, or the built-in default if not set. """ defaults: DefaultsMap = { "max_walk_commits": _DEFAULT_MAX_WALK_COMMITS, "max_ancestors": _DEFAULT_MAX_ANCESTORS, "max_graph_commits": _DEFAULT_MAX_GRAPH_COMMITS, "shard_prefix_length": _DEFAULT_SHARD_PREFIX_LENGTH, } default = defaults.get(key, 10_000) config = _load_config(_config_path(repo_root)) limits = config.get("limits") or {} # Explicit key dispatch keeps mypy happy on TypedDict literal-required keys. if key == "max_walk_commits": val: int | None = limits.get("max_walk_commits") elif key == "max_ancestors": val = limits.get("max_ancestors") elif key == "max_graph_commits": val = limits.get("max_graph_commits") elif key == "shard_prefix_length": val = limits.get("shard_prefix_length") else: val = None if isinstance(val, int) and val > 0: return val return default def get_config_value(key: str, repo_root: pathlib.Path | None = None) -> str | None: """Get a config value by dotted key (e.g. ``user.handle``, ``hub.url``). Returns ``None`` when the key is not set or the namespace is unknown. Args: key: Dotted key in ``.`` form. repo_root: Repository root. Defaults to ``Path.cwd()``. Returns: String value, or ``None``. """ parts = key.split(".", 1) if len(parts) != 2: return None namespace, subkey = parts config = _load_config(_config_path(repo_root)) if namespace == "user": # User identity lives in identity.toml, keyed by the configured hub URL. hub_url = (config.get("hub") or {}).get("url", "") if not hub_url: return None try: from muse.core.identity import load_identity, hostname_from_url hostname = hostname_from_url(hub_url) entry = load_identity(hostname) if entry is None: return None if subkey == "handle": return entry.get("handle") if subkey == "type": return entry.get("type") if subkey == "display_name": return entry.get("display_name") if subkey == "email": return entry.get("email") except Exception: pass return None if namespace == "hub": hub = config.get("hub") or {} if subkey == "url": return hub.get("url") return None if namespace == "domain": domain = config.get("domain") or {} return domain.get(subkey) if namespace == "limits": limits = config.get("limits") or {} if subkey == "max_walk_commits": v = limits.get("max_walk_commits") return str(v) if isinstance(v, int) else None if subkey == "max_ancestors": v = limits.get("max_ancestors") return str(v) if isinstance(v, int) else None if subkey == "max_graph_commits": v = limits.get("max_graph_commits") return str(v) if isinstance(v, int) else None if subkey == "shard_prefix_length": v = limits.get("shard_prefix_length") return str(v) if isinstance(v, int) else None return None if namespace == "commit": commit = config.get("commit") or {} if subkey == "sign": v = commit.get("sign") if v is True: return "true" if v is False: return "false" return None return None if namespace == "reflog": reflog = config.get("reflog") or {} if subkey == "expire-days": v = reflog.get("expire_days") return str(v) if isinstance(v, int) else None return None if namespace == "symlog": symlog = config.get("symlog") or {} if subkey == "expire-days": v = symlog.get("expire_days") return str(v) if isinstance(v, int) else None return None if namespace == "push": push = config.get("push") or {} if subkey in ("tags", "reflog", "symlog"): v = push.get(subkey) if v is True: return "true" if v is False: return "false" return None return None if namespace == "fetch": fetch = config.get("fetch") or {} if subkey == "prune": v = fetch.get("prune") if v is True: return "true" if v is False: return "false" return None return None return None def set_config_value(key: str, value: str, repo_root: pathlib.Path | None = None) -> None: """Set a config value by dotted key (e.g. ``user.handle``, ``domain.ticks_per_beat``). Args: key: Dotted key in ``.`` form. value: New string value. repo_root: Repository root. Defaults to ``Path.cwd()``. Raises: ValueError: If the namespace is blocked, unknown, or the subkey is invalid. """ parts = key.split(".", 1) if len(parts) != 2: raise ValueError(f"Key must be in 'namespace.subkey' form, got: {key!r}") namespace, subkey = parts if namespace in _BLOCKED_NAMESPACES: raise ValueError(_BLOCKED_NAMESPACES[namespace]) if namespace not in _SETTABLE_NAMESPACES: raise ValueError( f"Unknown config namespace {namespace!r}. " f"Settable namespaces: {', '.join(sorted(_SETTABLE_NAMESPACES))}" ) cp = _config_path(repo_root) cp.parent.mkdir(parents=True, exist_ok=True) config = _load_config(cp) if namespace == "user": set_user_field(subkey, value, repo_root) return if namespace == "hub": if subkey != "url": raise ValueError(f"Unknown [hub] config key: {subkey!r}. Valid keys: url") # Route through set_hub_url — it enforces the HTTPS requirement. set_hub_url(value, repo_root) return if namespace == "limits": _LIMITS_KEYS = frozenset({ "max_walk_commits", "max_ancestors", "max_graph_commits", "shard_prefix_length", }) if subkey not in _LIMITS_KEYS: raise ValueError( f"Unknown [limits] config key: {subkey!r}. " f"Valid keys: {', '.join(sorted(_LIMITS_KEYS))}" ) try: int_value = int(value) except ValueError as exc: raise ValueError( f"[limits] {subkey} must be an integer, got: {value!r}" ) from exc if int_value <= 0: raise ValueError(f"[limits] {subkey} must be a positive integer, got: {int_value}") if subkey == "shard_prefix_length" and int_value not in (2, 4): raise ValueError("shard_prefix_length must be 2 or 4") limits_section: LimitsConfig = config.get("limits") or {} if subkey == "max_walk_commits": limits_section["max_walk_commits"] = int_value elif subkey == "max_ancestors": limits_section["max_ancestors"] = int_value elif subkey == "max_graph_commits": limits_section["max_graph_commits"] = int_value elif subkey == "shard_prefix_length": limits_section["shard_prefix_length"] = int_value config["limits"] = limits_section write_text_atomic(cp, _dump_toml(config)) logger.info("✅ limits.%s = %d", subkey, int_value) return if namespace == "commit": _COMMIT_KEYS = frozenset({"sign"}) if subkey not in _COMMIT_KEYS: raise ValueError( f"Unknown [commit] config key: {subkey!r}. " f"Valid keys: {', '.join(sorted(_COMMIT_KEYS))}" ) if value not in ("true", "false"): raise ValueError(f"[commit] {subkey} must be 'true' or 'false', got: {value!r}") commit_section: CommitConfig = config.get("commit") or {} if subkey == "sign": commit_section["sign"] = value == "true" config["commit"] = commit_section write_text_atomic(cp, _dump_toml(config)) logger.info("✅ commit.%s = %s", subkey, value) return if namespace == "reflog": _REFLOG_KEYS = frozenset({"expire-days"}) if subkey not in _REFLOG_KEYS: raise ValueError( f"Unknown [reflog] config key: {subkey!r}. " f"Valid keys: {', '.join(sorted(_REFLOG_KEYS))}" ) if subkey == "expire-days": try: int_value = int(value) except ValueError as exc: raise ValueError( f"[reflog] expire-days must be a positive integer, got: {value!r}" ) from exc if int_value <= 0: raise ValueError(f"[reflog] expire-days must be a positive integer, got: {int_value}") reflog_section: dict = config.get("reflog") or {} reflog_section["expire_days"] = int_value config["reflog"] = reflog_section write_text_atomic(cp, _dump_toml(config)) logger.info("✅ reflog.expire-days = %d", int_value) return if namespace == "symlog": _SYMLOG_KEYS = frozenset({"expire-days"}) if subkey not in _SYMLOG_KEYS: raise ValueError( f"Unknown [symlog] config key: {subkey!r}. " f"Valid keys: {', '.join(sorted(_SYMLOG_KEYS))}" ) if subkey == "expire-days": try: sl_int_value = int(value) except ValueError as exc: raise ValueError( f"[symlog] expire-days must be a positive integer, got: {value!r}" ) from exc if sl_int_value <= 0: raise ValueError(f"[symlog] expire-days must be a positive integer, got: {sl_int_value}") symlog_section: SymlogConfig = config.get("symlog") or {} symlog_section["expire_days"] = sl_int_value config["symlog"] = symlog_section write_text_atomic(cp, _dump_toml(config)) logger.info("✅ symlog.expire-days = %d", sl_int_value) return if namespace == "push": _PUSH_KEYS = frozenset({"tags", "reflog", "symlog"}) if subkey not in _PUSH_KEYS: raise ValueError( f"Unknown [push] config key: {subkey!r}. " f"Valid keys: {', '.join(sorted(_PUSH_KEYS))}" ) if value not in ("true", "false"): raise ValueError(f"[push] {subkey} must be 'true' or 'false', got: {value!r}") push_section: dict = config.get("push") or {} push_section[subkey] = value == "true" config["push"] = push_section write_text_atomic(cp, _dump_toml(config)) logger.info("✅ push.%s = %s", subkey, value) return if namespace == "fetch": _FETCH_KEYS = frozenset({"prune"}) if subkey not in _FETCH_KEYS: raise ValueError( f"Unknown [fetch] config key: {subkey!r}. " f"Valid keys: {', '.join(sorted(_FETCH_KEYS))}" ) if value not in ("true", "false"): raise ValueError(f"[fetch] {subkey} must be 'true' or 'false', got: {value!r}") fetch_section: dict = config.get("fetch") or {} fetch_section[subkey] = value == "true" config["fetch"] = fetch_section write_text_atomic(cp, _dump_toml(config)) logger.info("✅ fetch.%s = %s", subkey, value) return # namespace == "domain" _validate_toml_key(subkey, "domain key") domain: DomainConfig = config.get("domain") or {} domain[subkey] = value config["domain"] = domain write_text_atomic(cp, _dump_toml(config)) logger.info("✅ domain.%s = %r", subkey, value) def config_as_dict(repo_root: pathlib.Path | None = None) -> ConfigTree: """Return the full config as a plain ``dict[str, dict[str, str]]`` for JSON output. Credentials are never included — the hub section only contains the URL. Args: repo_root: Repository root. Defaults to ``Path.cwd()``. Returns: Nested dict suitable for ``json.dumps``. """ config = _load_config(_config_path(repo_root)) result: ConfigTree = {} hub = config.get("hub") if hub: hub_url = hub.get("url", "") if hub_url: result["hub"] = {"url": hub_url} remotes = config.get("remotes") or {} if remotes: remotes_dict: ConfigSection = {} for rname, entry in sorted(remotes.items()): url = entry.get("url", "") if url: remotes_dict[rname] = url if remotes_dict: result["remotes"] = remotes_dict domain = config.get("domain") or {} if domain: result["domain"] = dict(sorted(domain.items())) limits = config.get("limits") or {} if limits: limits_dict: ConfigSection = {} for lk in ("max_walk_commits", "max_ancestors", "max_graph_commits", "shard_prefix_length"): lv = limits.get(lk) if lv is not None: limits_dict[lk] = str(lv) if limits_dict: result["limits"] = limits_dict return result def config_path_for_editor(repo_root: pathlib.Path | None = None) -> pathlib.Path: """Return the config path for the ``config edit`` command.""" return _config_path(repo_root) # --------------------------------------------------------------------------- # Branch metadata helpers # --------------------------------------------------------------------------- def write_branch_meta( repo_root: pathlib.Path, branch_name: str, *, intent: str | None = None, resumable: bool | None = None, remote: str | None = None, merge: str | None = None, created_by_handle: str | None = None, created_by_kind: str | None = None, created_at: str | None = None, ) -> None: """Write per-branch metadata to ``[branch.""]`` in ``.muse/config.toml``. Only the supplied keyword arguments are updated; existing fields (``remote``, ``merge``, and previously written ``intent``/``resumable``) are preserved unchanged. Args: repo_root: Repository root directory. branch_name: Name of the branch (e.g. ``"feat/my-thing"``). intent: Short description of what this branch is for. resumable: Mark this branch as a resumable agent checkpoint. remote: Upstream remote name (e.g. ``"origin"``) — musehub#217. merge: Upstream merge ref (e.g. ``"refs/heads/main"``) — musehub#217. created_by_handle: Who created this branch — musehub#216. Callers must only pass this at actual creation time; this function does not itself enforce write-once semantics. created_by_kind: ``"human"`` or ``"agent"`` — musehub#216. created_at: ISO 8601 creation timestamp — musehub#216. """ _validate_toml_key(branch_name, "branch name") cp = _config_path(repo_root) cp.parent.mkdir(parents=True, exist_ok=True) config = _load_config(cp) branch_map: dict[str, BranchMeta] = dict(config.get("branch") or {}) entry: BranchMeta = dict(branch_map.get(branch_name) or {}) # type: ignore[arg-type] if intent is not None: entry["intent"] = intent if resumable is not None: entry["resumable"] = resumable if remote is not None: entry["remote"] = remote if merge is not None: entry["merge"] = merge if created_by_handle is not None: entry["created_by_handle"] = created_by_handle if created_by_kind is not None: entry["created_by_kind"] = created_by_kind if created_at is not None: entry["created_at"] = created_at branch_map[branch_name] = entry config["branch"] = branch_map write_text_atomic(cp, _dump_toml(config)) def resolve_actor_identity(repo_root: pathlib.Path | None = None) -> tuple[str, str]: """Return ``(handle, kind)`` for whoever is currently invoking the CLI. Lightweight, local-only identity resolution — no key material, no hub round-trip — used to attribute branch-creation provenance (musehub#216). Resolution order: 1. ``MUSE_AGENT_HANDLE`` env var (set by agentception for spawned agents) — returns ``(handle, "agent")``. 2. ``user.handle`` from config — returns ``(handle, "human")``. 3. Falls back to ``("human", "human")`` when neither is set. Args: repo_root: Repository root. Defaults to ``Path.cwd()``. """ import os as _os agent_handle = _os.environ.get("MUSE_AGENT_HANDLE", "").strip() if agent_handle: return (agent_handle, "agent") handle = get_config_value("user.handle", repo_root) or "human" return (handle, "human") def delete_branch_meta(repo_root: pathlib.Path, branch_name: str) -> None: """Remove the ``[branch.""]`` section from ``.muse/config.toml``. Called by ``muse branch -d/-D`` after a successful branch deletion so stale intent/resumable entries do not accumulate indefinitely. No-op when the branch has no metadata or the config file is absent. """ cp = _config_path(repo_root) if not cp.exists(): return config = _load_config(cp) branch_map = dict(config.get("branch") or {}) if branch_name not in branch_map: return del branch_map[branch_name] config["branch"] = branch_map write_text_atomic(cp, _dump_toml(config)) def read_branch_meta( repo_root: pathlib.Path, branch_name: str, ) -> BranchMeta: """Return per-branch metadata from ``.muse/config.toml``. Returns an empty dict when the branch has no metadata or the config file is absent. Args: repo_root: Repository root directory. branch_name: Name of the branch (e.g. ``"feat/my-thing"``). Returns: Dict with any of: ``intent`` (str), ``resumable`` (bool), ``remote`` (str), ``merge`` (str). """ config = _load_config(_config_path(repo_root)) branch_map = config.get("branch") or {} return dict(branch_map.get(branch_name) or {}) # --------------------------------------------------------------------------- # Protected branches helpers # --------------------------------------------------------------------------- def get_protected_branches(repo_root: pathlib.Path | None = None) -> list[str]: """Return the list of protected branch patterns from ``[protected_branches]``. Returns an empty list when the section is absent or has no ``branches`` key. Args: repo_root: Repository root. Defaults to ``Path.cwd()``. """ config = _load_config(_config_path(repo_root)) return list(config.get("protected_branches") or []) def is_branch_protected(branch: str, patterns: list[str]) -> bool: """Return ``True`` if *branch* matches any pattern in *patterns*. Patterns are matched with :func:`fnmatch.fnmatch` (shell-style globs). Matching is case-sensitive, consistent with Python fnmatch behaviour. Args: branch: Branch name to test (e.g. ``"release/1.0"``). patterns: List of patterns from ``[protected_branches] branches``. """ return any(fnmatch.fnmatch(branch, p) for p in patterns) # --------------------------------------------------------------------------- # Remote helpers # --------------------------------------------------------------------------- def get_remote(name: str, repo_root: pathlib.Path | None = None) -> str | None: """Return the URL for remote *name*, or ``None`` when not configured. Args: name: Remote name (e.g. ``"origin"``). repo_root: Repository root. Defaults to ``Path.cwd()``. Returns: URL string, or ``None``. """ config = _load_config(_config_path(repo_root)) remotes = config.get("remotes") if remotes is None: return None entry = remotes.get(name) if entry is None: return None url = entry.get("url", "") return url.strip() if url.strip() else None def set_remote( name: str, url: str, repo_root: pathlib.Path | None = None, ) -> None: """Write ``[remotes.] url`` to ``.muse/config.toml``. Preserves all other sections. Creates the file if absent. Args: name: Remote name (e.g. ``"origin"``). url: Remote URL. repo_root: Repository root. Defaults to ``Path.cwd()``. """ cp = _config_path(repo_root) cp.parent.mkdir(parents=True, exist_ok=True) config = _load_config(cp) existing_remotes = config.get("remotes") remotes: RemotesMap = {} if existing_remotes: remotes.update(existing_remotes) existing_entry = remotes.get(name) entry: RemoteEntry = {} if existing_entry is not None: if "url" in existing_entry: entry["url"] = existing_entry["url"] if "branch" in existing_entry: entry["branch"] = existing_entry["branch"] entry["url"] = url remotes[name] = entry config["remotes"] = remotes write_text_atomic(cp, _dump_toml(config)) logger.info("✅ Remote %r set to %s", name, url) def remove_remote( name: str, repo_root: pathlib.Path | None = None, ) -> None: """Remove a named remote and its tracking refs. Args: name: Remote name to remove. repo_root: Repository root. Defaults to ``Path.cwd()``. Raises: KeyError: If *name* is not a configured remote. """ cp = _config_path(repo_root) config = _load_config(cp) remotes = config.get("remotes") if remotes is None or name not in remotes: raise KeyError(name) del remotes[name] config["remotes"] = remotes write_text_atomic(cp, _dump_toml(config)) logger.info("✅ Remote %r removed from config", name) root = (repo_root or pathlib.Path.cwd()).resolve() refs_dir = _remote_tracking_dir(root, name) if refs_dir.is_symlink(): # Refuse to rmtree a symlink — following a symlink placed by an # attacker could delete files outside the repository tree. logger.warning("⚠️ Skipping rmtree: remotes dir %s is a symlink", refs_dir) elif refs_dir.is_dir(): shutil.rmtree(refs_dir) logger.debug("✅ Removed tracking refs dir %s", refs_dir) def rename_remote( old_name: str, new_name: str, repo_root: pathlib.Path | None = None, ) -> None: """Rename a remote and move its tracking refs. Args: old_name: Current remote name. new_name: Desired new remote name. repo_root: Repository root. Defaults to ``Path.cwd()``. Raises: KeyError: If *old_name* is not a configured remote. ValueError: If *new_name* is already configured. """ cp = _config_path(repo_root) config = _load_config(cp) remotes = config.get("remotes") if remotes is None or old_name not in remotes: raise KeyError(old_name) if new_name in remotes: raise ValueError(new_name) remotes[new_name] = remotes.pop(old_name) config["remotes"] = remotes write_text_atomic(cp, _dump_toml(config)) logger.info("✅ Remote %r renamed to %r", old_name, new_name) root = (repo_root or pathlib.Path.cwd()).resolve() old_refs_dir = _remote_tracking_dir(root, old_name) new_refs_dir = _remote_tracking_dir(root, new_name) if old_refs_dir.is_dir(): old_refs_dir.rename(new_refs_dir) logger.debug("✅ Moved tracking refs dir %s → %s", old_refs_dir, new_refs_dir) def list_remotes(repo_root: pathlib.Path | None = None) -> list[RemoteConfig]: """Return all configured remotes sorted alphabetically by name. Args: repo_root: Repository root. Defaults to ``Path.cwd()``. Returns: List of ``{"name": str, "url": str}`` dicts. """ config = _load_config(_config_path(repo_root)) remotes = config.get("remotes") if remotes is None: return [] result: list[RemoteConfig] = [] for remote_name in sorted(remotes): entry = remotes[remote_name] url = entry.get("url", "") if not url.strip(): continue rc = RemoteConfig(name=remote_name, url=url.strip()) result.append(rc) return result # --------------------------------------------------------------------------- # Remote tracking-head helpers # --------------------------------------------------------------------------- def _remote_head_path( remote_name: str, branch: str, repo_root: pathlib.Path | None = None, ) -> pathlib.Path: """Return the path to the remote tracking pointer file.""" root = (repo_root or pathlib.Path.cwd()).resolve() return _remote_ref_path(root, remote_name, branch) def get_remote_head( remote_name: str, branch: str, repo_root: pathlib.Path | None = None, ) -> str | None: """Return the last-known remote commit ID for *remote_name*/*branch*. Returns ``None`` when the tracking pointer does not exist. Args: remote_name: Remote name (e.g. ``"origin"``). branch: Branch name (e.g. ``"main"``). repo_root: Repository root. Defaults to ``Path.cwd()``. Returns: Commit ID string, or ``None``. """ return read_ref(_remote_head_path(remote_name, branch, repo_root)) def set_remote_head( remote_name: str, branch: str, commit_id: str, repo_root: pathlib.Path | None = None, ) -> None: """Write the remote tracking pointer for *remote_name*/*branch*. Args: remote_name: Remote name (e.g. ``"origin"``). branch: Branch name. commit_id: Commit ID to record as the known remote HEAD. repo_root: Repository root. Defaults to ``Path.cwd()``. """ pointer = _remote_head_path(remote_name, branch, repo_root) write_text_atomic(pointer, commit_id) logger.debug("✅ Remote head %s/%s → %s", remote_name, branch, short_id(commit_id)) def delete_remote_head( remote_name: str, branch: str, repo_root: pathlib.Path | None = None, ) -> bool: """Remove the local remote-tracking pointer for *remote_name*/*branch*. Used after ``muse push --delete`` deletes the branch on the server, or when pruning stale tracking refs with ``muse branch -dr``. Args: remote_name: Remote name (e.g. ``"origin"``). branch: Branch name (e.g. ``"feat/my-thing"``). repo_root: Repository root. Defaults to ``Path.cwd()``. Returns: ``True`` if the pointer file existed and was removed, ``False`` if it was already absent (idempotent). """ pointer = _remote_head_path(remote_name, branch, repo_root) if not pointer.is_file(): return False pointer.unlink() # Remove now-empty parent directories (mirrors _cleanup_empty_dirs in branch.py). remotes_dir = pointer.parent while remotes_dir.name != remote_name: try: remotes_dir.rmdir() except OSError: break remotes_dir = remotes_dir.parent logger.debug("🗑 Remote tracking ref %s/%s removed", remote_name, branch) return True # --------------------------------------------------------------------------- # Upstream tracking helpers # --------------------------------------------------------------------------- def set_upstream( branch: str, remote_name: str, repo_root: pathlib.Path | None = None, ) -> None: """Record *remote_name* as the upstream remote for *branch*. Writes to two places: ``[remotes.]`` (consulted by :func:`get_upstream`'s reverse lookup — which remote tracks *branch* — used by ``muse pull``/``muse remote status``/``muse status``) and ``[branch.""]`` (consulted by ``muse branch --json``'s ``upstream`` field). Both must be written for upstream tracking to be visible everywhere it's read — see musehub#217. Args: branch: Local (and remote) branch name. remote_name: Remote name. repo_root: Repository root. Defaults to ``Path.cwd()``. """ cp = _config_path(repo_root) cp.parent.mkdir(parents=True, exist_ok=True) config = _load_config(cp) existing_remotes = config.get("remotes") remotes: RemotesMap = {} if existing_remotes: remotes.update(existing_remotes) existing_entry = remotes.get(remote_name) entry: RemoteEntry = {} if existing_entry is not None: if "url" in existing_entry: entry["url"] = existing_entry["url"] if "branch" in existing_entry: entry["branch"] = existing_entry["branch"] entry["branch"] = branch remotes[remote_name] = entry config["remotes"] = remotes write_text_atomic(cp, _dump_toml(config)) write_branch_meta( repo_root, branch, remote=remote_name, merge=f"refs/heads/{branch}" ) logger.info("✅ Upstream for branch %r set to %s/%r", branch, remote_name, branch) def get_upstream( branch: str, repo_root: pathlib.Path | None = None, ) -> str | None: """Return the configured upstream remote name for *branch*, or ``None``. Args: branch: Local branch name. repo_root: Repository root. Defaults to ``Path.cwd()``. Returns: Remote name string, or ``None``. """ config = _load_config(_config_path(repo_root)) remotes = config.get("remotes") if remotes is None: return None for rname, entry in remotes.items(): tracked = entry.get("branch", "") if tracked.strip() == branch: return rname return None def get_fetch_prune(repo_root: pathlib.Path | None = None) -> bool: """Return the effective ``fetch.prune`` setting — mirrors git's own precedence (muse#22). Checks the repo-level ``.muse/config.toml`` first; if ``fetch.prune`` is unset there, falls back to the user-global ``~/.muse/config.toml`` (set via ``muse config set --global fetch.prune true``). Defaults to ``False`` when neither is set — pruning stays an explicit, opt-in behaviour (``muse fetch --prune`` always remains available regardless). Args: repo_root: Repository root. Defaults to ``Path.cwd()``. Returns: ``True`` when either scope has ``fetch.prune = true``, repo taking priority over global. """ repo_value = get_config_value("fetch.prune", repo_root) if repo_value is not None: return repo_value == "true" global_value = get_config_value("fetch.prune", pathlib.Path.home()) return global_value == "true" # --------------------------------------------------------------------------- # Global user config — ~/.muse/config.toml (safe_dirs) # --------------------------------------------------------------------------- _GLOBAL_MUSE_DIR = _user_muse_dir() _GLOBAL_CONFIG_FILE = _user_config_toml_path() def _load_global_config() -> _SecurityConfig: """Load ``~/.muse/config.toml`` and return the ``[security]`` section. Returns a dict with key ``safe_dirs`` mapping to a list of path strings. Returns ``{"safe_dirs": []}`` when the file is absent or unparseable. """ if not _GLOBAL_CONFIG_FILE.is_file(): return {"safe_dirs": []} try: with _GLOBAL_CONFIG_FILE.open("rb") as fh: raw = tomllib.load(fh) except Exception as exc: # noqa: BLE001 logger.warning("⚠️ Failed to parse %s: %s", _GLOBAL_CONFIG_FILE, exc) return {"safe_dirs": []} security_raw = raw.get("security") if not isinstance(security_raw, dict): return {"safe_dirs": []} dirs_raw = security_raw.get("safe_dirs") if not isinstance(dirs_raw, list): return {"safe_dirs": []} safe: list[str] = [d for d in dirs_raw if isinstance(d, str) and d.strip()] return {"safe_dirs": safe} def _save_global_config(safe_dirs: list[str]) -> None: """Write ``[security] safe_dirs`` to ``~/.muse/config.toml``. Preserves any other sections that may exist in the file. """ import os _GLOBAL_MUSE_DIR.mkdir(parents=True, exist_ok=True) # Read existing raw content to preserve other sections. existing_lines: list[str] = [] if _GLOBAL_CONFIG_FILE.is_file(): try: existing_lines = _GLOBAL_CONFIG_FILE.read_text("utf-8").splitlines() except Exception: # noqa: BLE001 existing_lines = [] # Strip any existing [security] section from the file. filtered: list[str] = [] in_security = False for line in existing_lines: stripped = line.strip() if stripped == "[security]": in_security = True continue if in_security and stripped.startswith("["): in_security = False if not in_security: filtered.append(line) # Remove trailing blank lines before appending the new section. while filtered and not filtered[-1].strip(): filtered.pop() # Append the new [security] section. filtered.append("") filtered.append("[security]") if safe_dirs: items = ", ".join(f'"{_escape(d)}"' for d in safe_dirs) filtered.append(f"safe_dirs = [{items}]") else: filtered.append("safe_dirs = []") filtered.append("") content = "\n".join(filtered) tmp = _GLOBAL_CONFIG_FILE.with_suffix(".toml.tmp") tmp.write_text(content, encoding="utf-8") os.replace(tmp, _GLOBAL_CONFIG_FILE) def get_global_safe_dirs() -> list[str]: """Return the ``safe_dirs`` list from ``~/.muse/config.toml``. Returns an empty list when not configured. """ return _load_global_config().get("safe_dirs", []) def add_global_safe_dir(path: str) -> None: """Add *path* to the ``safe_dirs`` list in ``~/.muse/config.toml``. Normalises the path (``os.path.abspath``) before storing. Idempotent — adding the same path twice has no effect. Args: path: Absolute or relative path to trust. """ import os abs_path = os.path.abspath(path) current = get_global_safe_dirs() if abs_path not in current: current.append(abs_path) _save_global_config(current) logger.info("✅ Trusted path added: %s", abs_path) def remove_global_safe_dir(path: str) -> None: """Remove *path* from the ``safe_dirs`` list in ``~/.muse/config.toml``. Normalises the path before matching. No-op when the path is not present. Args: path: Absolute or relative path to remove from the trust list. """ import os abs_path = os.path.abspath(path) current = get_global_safe_dirs() updated = [d for d in current if d != abs_path] _save_global_config(updated) logger.info("✅ Trusted path removed: %s", abs_path)