aaronrene / gabriel-muse public
config.py python
1,270 lines 42.4 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 32 days ago
1 """Muse CLI configuration helpers.
2
3 Reads and writes ``.muse/config.toml`` — the per-repository configuration
4 file. Credentials (signing identities) are **not** stored here; they live in
5 ``~/.muse/identity.toml`` managed by :mod:`muse.core.identity`.
6
7 Config schema
8 -------------
9 ::
10
11 [user]
12 name = "Alice" # display name (human or agent handle)
13 email = "[email protected]"
14 type = "human" # "human" | "agent"
15
16 [hub]
17 url = "https://musehub.ai" # MuseHub fabric endpoint for this repo
18
19 [remotes.origin]
20 url = "https://hub.muse.io/repos/my-repo"
21 branch = "main"
22
23 [domain]
24 # Domain-specific key/value pairs; read by the active domain plugin.
25 # ticks_per_beat = "480"
26
27 Settable via ``muse config set``
28 ---------------------------------
29 - ``user.name``, ``user.email``, ``user.type``
30 - ``hub.url`` (alias: ``muse hub connect <url>``)
31 - ``domain.*``
32
33 Not settable via ``muse config set``
34 --------------------------------------
35 - ``remotes.*`` — use ``muse remote add/remove``
36 - credentials — use ``muse auth register``
37
38 Token resolution
39 ----------------
40 :func:`get_signing_identity` reads the hub URL from this file, then resolves the
41 signing identity from ``~/.muse/identity.toml`` via
42 :func:`muse.core.identity.resolve_token`. The token is **never** logged.
43 """
44
45 from __future__ import annotations
46
47 import logging
48 import pathlib
49 import shutil
50 import subprocess
51 import tomllib
52 from typing import TypedDict
53
54 from muse.core.store import write_text_atomic
55
56 logger = logging.getLogger(__name__)
57
58 type RemotesMap = dict[str, RemoteEntry] # remote_name → remote entry
59 type DomainConfig = dict[str, str] # domain key → value
60 type ConfigSection = dict[str, str] # generic flattened section dict
61 type ConfigTree = dict[str, dict[str, str]] # section → key → value (for JSON output)
62 type DefaultsMap = dict[str, int] # config key → default int value
63
64 _CONFIG_FILENAME = "config.toml"
65 _MUSE_DIR = ".muse"
66
67
68 # ---------------------------------------------------------------------------
69 # Named configuration types
70 # ---------------------------------------------------------------------------
71
72
73 class UserConfig(TypedDict, total=False):
74 """``[user]`` section in ``.muse/config.toml``."""
75
76 name: str
77 email: str
78 type: str # "human" | "agent"
79
80
81 class HubConfig(TypedDict, total=False):
82 """``[hub]`` section in ``.muse/config.toml``."""
83
84 url: str
85
86
87 class RemoteEntry(TypedDict, total=False):
88 """``[remotes.<name>]`` section in ``.muse/config.toml``."""
89
90 url: str
91 branch: str
92 pack_origin: str
93
94
95 class LimitsConfig(TypedDict, total=False):
96 """``[limits]`` section in ``.muse/config.toml``.
97
98 All values are optional — defaults are used when absent. Keys map to
99 the ``[limits]`` TOML table::
100
101 [limits]
102 max_walk_commits = 10000 # cap for walk_commits_between / muse log
103 max_ancestors = 50000 # cap for find_merge_base BFS
104 max_graph_commits = 50000 # cap for _collect_all_commits (--graph --all)
105 shard_prefix_length = 2 # object store shard depth: 2 (256 shards)
106 # or 4 (65536 shards) for very large repos
107 """
108
109 max_walk_commits: int
110 max_ancestors: int
111 max_graph_commits: int
112 shard_prefix_length: int
113
114
115 class MuseConfig(TypedDict, total=False):
116 """Structured view of the entire ``.muse/config.toml`` file."""
117
118 user: UserConfig
119 hub: HubConfig
120 remotes: RemotesMap
121 domain: DomainConfig
122 limits: LimitsConfig
123
124
125 class RemoteConfig(TypedDict, total=False):
126 """Public-facing remote descriptor returned by :func:`list_remotes`.
127
128 ``pack_origin`` is optional. When set, pack uploads are sent to
129 ``{pack_origin}/{owner}/{slug}/push/object-pack`` instead of the default
130 ``{url}/push/object-pack``. Use this to route small-object packs through
131 a Cloudflare Worker (or equivalent edge cache) without changing the primary
132 remote URL.
133 """
134
135 name: str # always present
136 url: str # always present
137 pack_origin: str # optional — overrides the origin for pack uploads
138
139
140 # ---------------------------------------------------------------------------
141 # Internal helpers
142 # ---------------------------------------------------------------------------
143
144
145 def _config_path(repo_root: pathlib.Path | None) -> pathlib.Path:
146 """Return the path to .muse/config.toml for the given (or cwd) root."""
147 root = (repo_root or pathlib.Path.cwd()).resolve()
148 return root / _MUSE_DIR / _CONFIG_FILENAME
149
150
151 def _load_config(config_path: pathlib.Path) -> MuseConfig:
152 """Load and parse config.toml; return an empty MuseConfig if absent."""
153 if not config_path.is_file():
154 return {}
155
156 try:
157 with config_path.open("rb") as fh:
158 raw = tomllib.load(fh)
159 except Exception as exc: # noqa: BLE001
160 logger.warning("⚠️ Failed to parse %s: %s", config_path, exc)
161 return {}
162
163 config: MuseConfig = {}
164
165 user_raw = raw.get("user")
166 if isinstance(user_raw, dict):
167 user: UserConfig = {}
168 name_val = user_raw.get("name")
169 if isinstance(name_val, str):
170 user["name"] = name_val
171 email_val = user_raw.get("email")
172 if isinstance(email_val, str):
173 user["email"] = email_val
174 type_val = user_raw.get("type")
175 if isinstance(type_val, str):
176 user["type"] = type_val
177 config["user"] = user
178
179 hub_raw = raw.get("hub")
180 if isinstance(hub_raw, dict):
181 hub: HubConfig = {}
182 url_val = hub_raw.get("url")
183 if isinstance(url_val, str):
184 hub["url"] = url_val
185 config["hub"] = hub
186
187 remotes_raw = raw.get("remotes")
188 if isinstance(remotes_raw, dict):
189 remotes: RemotesMap = {}
190 for name, remote_raw in remotes_raw.items():
191 if isinstance(remote_raw, dict):
192 entry: RemoteEntry = {}
193 rurl = remote_raw.get("url")
194 if isinstance(rurl, str):
195 entry["url"] = rurl
196 branch_val = remote_raw.get("branch")
197 if isinstance(branch_val, str):
198 entry["branch"] = branch_val
199 pack_origin_val = remote_raw.get("pack_origin")
200 if isinstance(pack_origin_val, str):
201 entry["pack_origin"] = pack_origin_val
202 remotes[name] = entry
203 config["remotes"] = remotes
204
205 domain_raw = raw.get("domain")
206 if isinstance(domain_raw, dict):
207 domain: DomainConfig = {}
208 for key, val in domain_raw.items():
209 if isinstance(val, str):
210 domain[key] = val
211 config["domain"] = domain
212
213 limits_raw = raw.get("limits")
214 if isinstance(limits_raw, dict):
215 limits: LimitsConfig = {}
216 mwc = limits_raw.get("max_walk_commits")
217 if isinstance(mwc, int) and mwc > 0:
218 limits["max_walk_commits"] = mwc
219 ma = limits_raw.get("max_ancestors")
220 if isinstance(ma, int) and ma > 0:
221 limits["max_ancestors"] = ma
222 mgc = limits_raw.get("max_graph_commits")
223 if isinstance(mgc, int) and mgc > 0:
224 limits["max_graph_commits"] = mgc
225 spl = limits_raw.get("shard_prefix_length")
226 if isinstance(spl, int) and spl in (2, 4):
227 limits["shard_prefix_length"] = spl
228 config["limits"] = limits
229
230 return config
231
232
233 def _escape(value: str) -> str:
234 """Escape a TOML basic string value (backslash and double-quote only).
235
236 TOML basic strings allow control characters escaped as ``\\n``, ``\\t``,
237 etc., but we store only printable content — control characters in values
238 are also stripped here so the resulting TOML file remains parseable.
239 """
240 return (
241 value.replace("\\", "\\\\")
242 .replace('"', '\\"')
243 .replace("\n", "\\n")
244 .replace("\r", "\\r")
245 .replace("\0", "")
246 )
247
248
249 # Characters that are structurally significant in unquoted TOML keys and
250 # table headers. Any of these in a key name would allow injection of
251 # arbitrary TOML sections or key-value pairs.
252 _TOML_KEY_UNSAFE: frozenset[str] = frozenset('\n\r\0][="')
253
254
255 def _validate_toml_key(key: str, context: str = "key") -> None:
256 """Raise ``ValueError`` if *key* contains TOML-structurally unsafe characters.
257
258 Prevents injection attacks where a crafted key like ``x]\\n[evil`` would
259 break the TOML section structure and allow writing arbitrary sections.
260
261 Args:
262 key: Key string to validate.
263 context: Human-readable label used in the error message (e.g. ``"domain key"``).
264
265 Raises:
266 ValueError: If any character in *key* is in ``_TOML_KEY_UNSAFE``.
267 """
268 bad = _TOML_KEY_UNSAFE & set(key)
269 if bad:
270 chars = ", ".join(sorted(repr(c) for c in bad))
271 raise ValueError(
272 f"Config {context} {key!r} contains characters not allowed in TOML keys: {chars}"
273 )
274
275
276 def _dump_toml(config: MuseConfig) -> str:
277 """Serialise a MuseConfig to TOML text.
278
279 Section order: ``[user]``, ``[hub]``, ``[remotes.*]``, ``[domain]``, ``[limits]``.
280
281 All key names are validated against ``_TOML_KEY_UNSAFE`` before being
282 written, preventing TOML injection via crafted domain keys or remote names.
283 """
284 lines: list[str] = []
285
286 user = config.get("user")
287 if user:
288 lines.append("[user]")
289 name = user.get("name", "")
290 if name:
291 lines.append(f'name = "{_escape(name)}"')
292 email = user.get("email", "")
293 if email:
294 lines.append(f'email = "{_escape(email)}"')
295 utype = user.get("type", "")
296 if utype:
297 lines.append(f'type = "{_escape(utype)}"')
298 lines.append("")
299
300 hub = config.get("hub")
301 if hub:
302 lines.append("[hub]")
303 url = hub.get("url", "")
304 if url:
305 lines.append(f'url = "{_escape(url)}"')
306 lines.append("")
307
308 remotes = config.get("remotes") or {}
309 for remote_name in sorted(remotes):
310 # Remote names come from _load_config which parses TOML, so they are
311 # safe at read time. Validate defensively before writing.
312 _validate_toml_key(remote_name, "remote name")
313 entry = remotes[remote_name]
314 lines.append(f"[remotes.{remote_name}]")
315 rurl = entry.get("url", "")
316 if rurl:
317 lines.append(f'url = "{_escape(rurl)}"')
318 branch = entry.get("branch", "")
319 if branch:
320 lines.append(f'branch = "{_escape(branch)}"')
321 lines.append("")
322
323 domain = config.get("domain") or {}
324 if domain:
325 lines.append("[domain]")
326 for key, val in sorted(domain.items()):
327 _validate_toml_key(key, "domain key")
328 lines.append(f'{key} = "{_escape(val)}"')
329 lines.append("")
330
331 limits = config.get("limits") or {}
332 if limits:
333 lines.append("[limits]")
334 mwc = limits.get("max_walk_commits")
335 if mwc is not None:
336 lines.append(f"max_walk_commits = {mwc}")
337 ma = limits.get("max_ancestors")
338 if ma is not None:
339 lines.append(f"max_ancestors = {ma}")
340 mgc = limits.get("max_graph_commits")
341 if mgc is not None:
342 lines.append(f"max_graph_commits = {mgc}")
343 spl = limits.get("shard_prefix_length")
344 if spl is not None:
345 lines.append(f"shard_prefix_length = {spl}")
346 lines.append("")
347
348 return "\n".join(lines)
349
350
351 # ---------------------------------------------------------------------------
352 # Auth token resolution (via identity store)
353 # ---------------------------------------------------------------------------
354
355
356 def get_signing_identity(
357 repo_root: pathlib.Path | None = None,
358 remote_url: str | None = None,
359 agent_id: str | None = None,
360 ) -> "object | None":
361 """Return a :class:`~muse.core.transport.SigningIdentity` for a hub, or ``None``.
362
363 Resolution order:
364 1. ``MUSE_AGENT_KEY`` environment variable — PEM private key bytes. When
365 set the key is used directly, bypassing the identity store entirely.
366 The handle is taken from ``MUSE_AGENT_HANDLE`` (defaults to *agent_id*
367 if that is also set, otherwise ``"agent"``). This is the injection
368 mechanism for agent subprocesses spawned by agentception.
369 2. Agent-specific entry in ``~/.muse/identity.toml`` keyed by
370 ``"hostname#agent_id"`` — when *agent_id* is provided.
371 3. Human entry in ``~/.muse/identity.toml`` keyed by bare hostname.
372 4. Hub URL from ``[hub] url`` in ``.muse/config.toml`` (fallback lookup
373 URL when *remote_url* is not supplied).
374
375 The private key is **never** logged.
376
377 Args:
378 repo_root: Repository root. Defaults to ``Path.cwd()``.
379 remote_url: URL of the specific remote being contacted.
380 agent_id: Agent handle, e.g. ``"agentception-abc123"``. Used to
381 try an agent-specific key before falling back to the
382 human key.
383
384 Returns:
385 :class:`~muse.core.transport.SigningIdentity` or ``None``.
386 """
387 import os as _os
388 from muse.core.identity import resolve_signing_identity # avoid circular import
389 from muse.core.transport import SigningIdentity
390
391 # 1. MUSE_AGENT_KEY env var — injected by agentception at spawn time.
392 agent_pem = _os.environ.get("MUSE_AGENT_KEY", "").strip()
393 if agent_pem:
394 from muse.core.keypair import load_private_key_from_pem
395 private_key = load_private_key_from_pem(agent_pem.encode())
396 if private_key is not None:
397 handle = (
398 _os.environ.get("MUSE_AGENT_HANDLE", "").strip()
399 or agent_id
400 or "agent"
401 )
402 logger.debug("✅ Signing identity from MUSE_AGENT_KEY (handle=%s)", handle)
403 return SigningIdentity(handle=handle, private_key=private_key)
404 logger.warning("⚠️ MUSE_AGENT_KEY is set but could not be decoded — ignoring")
405
406 # 2 & 3. Identity store lookup (agent key → human key fallback).
407 lookup_url: str | None = remote_url or get_hub_url(repo_root)
408 if lookup_url is None:
409 logger.debug("⚠️ No hub configured — skipping signing identity lookup")
410 return None
411
412 result = resolve_signing_identity(lookup_url, agent_id=agent_id)
413 if result is None:
414 logger.debug(
415 "⚠️ No signing identity for hub %s — run `muse auth keygen && muse auth register`",
416 lookup_url,
417 )
418 return None
419
420 handle, private_key = result
421 logger.debug("✅ Signing identity resolved for hub %s (handle=%s)", lookup_url, handle)
422 return SigningIdentity(handle=handle, private_key=private_key)
423
424
425
426
427 # ---------------------------------------------------------------------------
428 # Hub helpers
429 # ---------------------------------------------------------------------------
430
431
432 def get_hub_url(repo_root: pathlib.Path | None = None) -> str | None:
433 """Return the hub URL from ``[hub] url``, or ``None`` if not configured.
434
435 Resolution order:
436 1. ``<repo>/.muse/config.toml`` — repo-local config (highest priority).
437 2. ``~/.muse/config.toml`` — global user config (fallback).
438
439 This fallback ensures ``muse auth whoami`` and other hub-aware commands
440 work without ``--hub`` even when invoked outside a repository, as long as
441 the user has set a default hub in their global config.
442
443 Args:
444 repo_root: Repository root. Defaults to ``Path.cwd()``.
445
446 Returns:
447 URL string, or ``None``.
448 """
449 config = _load_config(_config_path(repo_root))
450 hub = config.get("hub")
451 if hub is not None:
452 url = hub.get("url", "")
453 if url.strip():
454 return url.strip()
455
456 # Fall back to ~/.muse/config.toml so hub-aware commands (e.g. `muse auth
457 # whoami`) work without --hub when invoked outside a repository.
458 global_config = _load_config(_GLOBAL_CONFIG_FILE)
459 global_hub = global_config.get("hub")
460 if global_hub is not None:
461 url = global_hub.get("url", "")
462 if url.strip():
463 return url.strip()
464
465 return None
466
467
468 def set_hub_url(url: str, repo_root: pathlib.Path | None = None) -> None:
469 """Write ``[hub] url`` to ``.muse/config.toml``.
470
471 Preserves all other sections. Creates the config file if absent.
472 Rejects ``http://`` URLs — Muse never contacts a hub over cleartext HTTP.
473
474 Args:
475 url: Hub URL (must be ``https://``).
476 repo_root: Repository root. Defaults to ``Path.cwd()``.
477
478 Raises:
479 ValueError: If *url* does not use the ``https://`` scheme.
480 """
481 _is_loopback = url.startswith("http://localhost") or url.startswith("http://127.0.0.1") or url.startswith("http://[::1]")
482 if not url.startswith("https://") and not _is_loopback:
483 raise ValueError(
484 f"Hub URL must use HTTPS. Got: {url!r}\n"
485 "Muse never connects to a hub over cleartext HTTP.\n"
486 "(Exception: http://localhost and http://127.0.0.1 are allowed for local development.)"
487 )
488 cp = _config_path(repo_root)
489 cp.parent.mkdir(parents=True, exist_ok=True)
490 config = _load_config(cp)
491 config["hub"] = HubConfig(url=url)
492 write_text_atomic(cp, _dump_toml(config))
493 logger.info("✅ Hub URL set to %s", url)
494
495
496 def clear_hub_url(repo_root: pathlib.Path | None = None) -> None:
497 """Remove the ``[hub]`` section from ``.muse/config.toml``.
498
499 Args:
500 repo_root: Repository root. Defaults to ``Path.cwd()``.
501 """
502 cp = _config_path(repo_root)
503 config = _load_config(cp)
504 if "hub" in config:
505 del config["hub"]
506 write_text_atomic(cp, _dump_toml(config))
507 logger.info("✅ Hub disconnected")
508
509
510 # ---------------------------------------------------------------------------
511 # User config helpers
512 # ---------------------------------------------------------------------------
513
514
515
516 def set_user_field(key: str, value: str, repo_root: pathlib.Path | None = None) -> None:
517 """Set a single ``[user]`` field by name.
518
519 Allowed keys: ``name``, ``email``, ``type``.
520
521 Args:
522 key: Field name within ``[user]``.
523 value: New value.
524 repo_root: Repository root. Defaults to ``Path.cwd()``.
525
526 Raises:
527 ValueError: If *key* is not a recognised user config field.
528 """
529 if key not in {"name", "email", "type"}:
530 raise ValueError(f"Unknown [user] config key: {key!r}. Valid keys: name, email, type")
531 cp = _config_path(repo_root)
532 cp.parent.mkdir(parents=True, exist_ok=True)
533 config = _load_config(cp)
534 user: UserConfig = config.get("user") or {}
535 if key == "name":
536 user["name"] = value
537 elif key == "email":
538 user["email"] = value
539 elif key == "type":
540 user["type"] = value
541 config["user"] = user
542 write_text_atomic(cp, _dump_toml(config))
543 logger.info("✅ user.%s = %r", key, value)
544
545
546 # ---------------------------------------------------------------------------
547 # Generic dotted-key helpers
548 # ---------------------------------------------------------------------------
549
550 _BlockedNS = dict[str, str]
551 _BLOCKED_NAMESPACES: _BlockedNS = {
552 "auth": "Use `muse auth keygen` and `muse auth register` to manage credentials.",
553 "remotes": "Use `muse remote add/remove/rename` to manage remotes.",
554 }
555
556 _SETTABLE_NAMESPACES = {"user", "hub", "domain", "limits"}
557
558 # Default cap values — used when the [limits] section is absent or the key
559 # is not set. These are the same values that were previously hardcoded inside
560 # the individual functions.
561 _DEFAULT_MAX_WALK_COMMITS: int = 10_000
562 _DEFAULT_MAX_ANCESTORS: int = 50_000
563 _DEFAULT_MAX_GRAPH_COMMITS: int = 50_000
564 _DEFAULT_SHARD_PREFIX_LENGTH: int = 2
565
566
567 def get_limit(key: str, repo_root: pathlib.Path | None = None) -> int:
568 """Return a ``[limits]`` integer cap from config, or its default.
569
570 Args:
571 key: Limit key — one of ``max_walk_commits``, ``max_ancestors``,
572 ``max_graph_commits``.
573 repo_root: Repository root; ``None`` falls back to ``Path.cwd()``.
574
575 Returns:
576 Configured integer value, or the built-in default if not set.
577 """
578 defaults: DefaultsMap = {
579 "max_walk_commits": _DEFAULT_MAX_WALK_COMMITS,
580 "max_ancestors": _DEFAULT_MAX_ANCESTORS,
581 "max_graph_commits": _DEFAULT_MAX_GRAPH_COMMITS,
582 "shard_prefix_length": _DEFAULT_SHARD_PREFIX_LENGTH,
583 }
584 default = defaults.get(key, 10_000)
585 config = _load_config(_config_path(repo_root))
586 limits = config.get("limits") or {}
587 # Explicit key dispatch keeps mypy happy on TypedDict literal-required keys.
588 if key == "max_walk_commits":
589 val: int | None = limits.get("max_walk_commits")
590 elif key == "max_ancestors":
591 val = limits.get("max_ancestors")
592 elif key == "max_graph_commits":
593 val = limits.get("max_graph_commits")
594 elif key == "shard_prefix_length":
595 val = limits.get("shard_prefix_length")
596 else:
597 val = None
598 if isinstance(val, int) and val > 0:
599 return val
600 return default
601
602
603 def get_config_value(key: str, repo_root: pathlib.Path | None = None) -> str | None:
604 """Get a config value by dotted key (e.g. ``user.name``, ``hub.url``).
605
606 Returns ``None`` when the key is not set or the namespace is unknown.
607
608 Args:
609 key: Dotted key in ``<namespace>.<subkey>`` form.
610 repo_root: Repository root. Defaults to ``Path.cwd()``.
611
612 Returns:
613 String value, or ``None``.
614 """
615 parts = key.split(".", 1)
616 if len(parts) != 2:
617 return None
618 namespace, subkey = parts
619 config = _load_config(_config_path(repo_root))
620
621 if namespace == "user":
622 user = config.get("user") or {}
623 if subkey == "name":
624 return user.get("name")
625 if subkey == "email":
626 return user.get("email")
627 if subkey == "type":
628 return user.get("type")
629 return None
630
631 if namespace == "hub":
632 hub = config.get("hub") or {}
633 if subkey == "url":
634 return hub.get("url")
635 return None
636
637 if namespace == "domain":
638 domain = config.get("domain") or {}
639 return domain.get(subkey)
640
641 if namespace == "limits":
642 limits = config.get("limits") or {}
643 if subkey == "max_walk_commits":
644 v = limits.get("max_walk_commits")
645 return str(v) if isinstance(v, int) else None
646 if subkey == "max_ancestors":
647 v = limits.get("max_ancestors")
648 return str(v) if isinstance(v, int) else None
649 if subkey == "max_graph_commits":
650 v = limits.get("max_graph_commits")
651 return str(v) if isinstance(v, int) else None
652 if subkey == "shard_prefix_length":
653 v = limits.get("shard_prefix_length")
654 return str(v) if isinstance(v, int) else None
655 return None
656
657 return None
658
659
660 def set_config_value(key: str, value: str, repo_root: pathlib.Path | None = None) -> None:
661 """Set a config value by dotted key (e.g. ``user.name``, ``domain.ticks_per_beat``).
662
663 Args:
664 key: Dotted key in ``<namespace>.<subkey>`` form.
665 value: New string value.
666 repo_root: Repository root. Defaults to ``Path.cwd()``.
667
668 Raises:
669 ValueError: If the namespace is blocked, unknown, or the subkey is invalid.
670 """
671 parts = key.split(".", 1)
672 if len(parts) != 2:
673 raise ValueError(f"Key must be in 'namespace.subkey' form, got: {key!r}")
674 namespace, subkey = parts
675
676 if namespace in _BLOCKED_NAMESPACES:
677 raise ValueError(_BLOCKED_NAMESPACES[namespace])
678
679 if namespace not in _SETTABLE_NAMESPACES:
680 raise ValueError(
681 f"Unknown config namespace {namespace!r}. "
682 f"Settable namespaces: {', '.join(sorted(_SETTABLE_NAMESPACES))}"
683 )
684
685 cp = _config_path(repo_root)
686 cp.parent.mkdir(parents=True, exist_ok=True)
687 config = _load_config(cp)
688
689 if namespace == "user":
690 set_user_field(subkey, value, repo_root)
691 return
692
693 if namespace == "hub":
694 if subkey != "url":
695 raise ValueError(f"Unknown [hub] config key: {subkey!r}. Valid keys: url")
696 # Route through set_hub_url — it enforces the HTTPS requirement.
697 set_hub_url(value, repo_root)
698 return
699
700 if namespace == "limits":
701 _LIMITS_KEYS = frozenset({
702 "max_walk_commits", "max_ancestors", "max_graph_commits", "shard_prefix_length",
703 })
704 if subkey not in _LIMITS_KEYS:
705 raise ValueError(
706 f"Unknown [limits] config key: {subkey!r}. "
707 f"Valid keys: {', '.join(sorted(_LIMITS_KEYS))}"
708 )
709 try:
710 int_value = int(value)
711 except ValueError as exc:
712 raise ValueError(
713 f"[limits] {subkey} must be an integer, got: {value!r}"
714 ) from exc
715 if int_value <= 0:
716 raise ValueError(f"[limits] {subkey} must be a positive integer, got: {int_value}")
717 if subkey == "shard_prefix_length" and int_value not in (2, 4):
718 raise ValueError("shard_prefix_length must be 2 or 4")
719 limits_section: LimitsConfig = config.get("limits") or {}
720 if subkey == "max_walk_commits":
721 limits_section["max_walk_commits"] = int_value
722 elif subkey == "max_ancestors":
723 limits_section["max_ancestors"] = int_value
724 elif subkey == "max_graph_commits":
725 limits_section["max_graph_commits"] = int_value
726 elif subkey == "shard_prefix_length":
727 limits_section["shard_prefix_length"] = int_value
728 config["limits"] = limits_section
729 write_text_atomic(cp, _dump_toml(config))
730 logger.info("✅ limits.%s = %d", subkey, int_value)
731 return
732
733 # namespace == "domain"
734 _validate_toml_key(subkey, "domain key")
735 domain: DomainConfig = config.get("domain") or {}
736 domain[subkey] = value
737 config["domain"] = domain
738 write_text_atomic(cp, _dump_toml(config))
739 logger.info("✅ domain.%s = %r", subkey, value)
740
741
742 def config_as_dict(repo_root: pathlib.Path | None = None) -> ConfigTree:
743 """Return the full config as a plain ``dict[str, dict[str, str]]`` for JSON output.
744
745 Credentials are never included — the hub section only contains the URL.
746
747 Args:
748 repo_root: Repository root. Defaults to ``Path.cwd()``.
749
750 Returns:
751 Nested dict suitable for ``json.dumps``.
752 """
753 config = _load_config(_config_path(repo_root))
754 result: ConfigTree = {}
755
756 user = config.get("user")
757 if user:
758 user_dict: ConfigSection = {}
759 uname = user.get("name")
760 if uname:
761 user_dict["name"] = uname
762 uemail = user.get("email")
763 if uemail:
764 user_dict["email"] = uemail
765 utype = user.get("type")
766 if utype:
767 user_dict["type"] = utype
768 if user_dict:
769 result["user"] = user_dict
770
771 hub = config.get("hub")
772 if hub:
773 hub_url = hub.get("url", "")
774 if hub_url:
775 result["hub"] = {"url": hub_url}
776
777 remotes = config.get("remotes") or {}
778 if remotes:
779 remotes_dict: ConfigSection = {}
780 for rname, entry in sorted(remotes.items()):
781 url = entry.get("url", "")
782 if url:
783 remotes_dict[rname] = url
784 if remotes_dict:
785 result["remotes"] = remotes_dict
786
787 domain = config.get("domain") or {}
788 if domain:
789 result["domain"] = dict(sorted(domain.items()))
790
791 limits = config.get("limits") or {}
792 if limits:
793 limits_dict: ConfigSection = {}
794 for lk in ("max_walk_commits", "max_ancestors", "max_graph_commits", "shard_prefix_length"):
795 lv = limits.get(lk)
796 if lv is not None:
797 limits_dict[lk] = str(lv)
798 if limits_dict:
799 result["limits"] = limits_dict
800
801 return result
802
803
804 def config_path_for_editor(repo_root: pathlib.Path | None = None) -> pathlib.Path:
805 """Return the config path for the ``config edit`` command."""
806 return _config_path(repo_root)
807
808
809 # ---------------------------------------------------------------------------
810 # Remote helpers
811 # ---------------------------------------------------------------------------
812
813
814 def get_remote(name: str, repo_root: pathlib.Path | None = None) -> str | None:
815 """Return the URL for remote *name*, or ``None`` when not configured.
816
817 Args:
818 name: Remote name (e.g. ``"origin"``).
819 repo_root: Repository root. Defaults to ``Path.cwd()``.
820
821 Returns:
822 URL string, or ``None``.
823 """
824 config = _load_config(_config_path(repo_root))
825 remotes = config.get("remotes")
826 if remotes is None:
827 return None
828 entry = remotes.get(name)
829 if entry is None:
830 return None
831 url = entry.get("url", "")
832 return url.strip() if url.strip() else None
833
834
835 def get_remote_pack_origin(
836 name: str,
837 repo_root: pathlib.Path | None = None,
838 ) -> str | None:
839 """Return the ``pack_origin`` override for remote *name*, or ``None``.
840
841 When set, pack uploads (``POST /push/object-pack``) are directed to this
842 origin instead of the primary remote URL. The path component (owner/slug)
843 is preserved from the primary URL so the same MWP path structure applies.
844
845 Args:
846 name: Remote name (e.g. ``"local"``).
847 repo_root: Repository root. Defaults to ``Path.cwd()``.
848
849 Returns:
850 ``pack_origin`` string, or ``None`` when not configured.
851 """
852 config = _load_config(_config_path(repo_root))
853 remotes = config.get("remotes")
854 if remotes is None:
855 return None
856 entry = remotes.get(name)
857 if entry is None:
858 return None
859 origin = entry.get("pack_origin", "")
860 return origin.strip() if origin.strip() else None
861
862
863 def set_remote(
864 name: str,
865 url: str,
866 repo_root: pathlib.Path | None = None,
867 ) -> None:
868 """Write ``[remotes.<name>] url`` to ``.muse/config.toml``.
869
870 Preserves all other sections. Creates the file if absent.
871
872 Args:
873 name: Remote name (e.g. ``"origin"``).
874 url: Remote URL.
875 repo_root: Repository root. Defaults to ``Path.cwd()``.
876 """
877 cp = _config_path(repo_root)
878 cp.parent.mkdir(parents=True, exist_ok=True)
879 config = _load_config(cp)
880 existing_remotes = config.get("remotes")
881 remotes: RemotesMap = {}
882 if existing_remotes:
883 remotes.update(existing_remotes)
884 existing_entry = remotes.get(name)
885 entry: RemoteEntry = {}
886 if existing_entry is not None:
887 if "url" in existing_entry:
888 entry["url"] = existing_entry["url"]
889 if "branch" in existing_entry:
890 entry["branch"] = existing_entry["branch"]
891 entry["url"] = url
892 remotes[name] = entry
893 config["remotes"] = remotes
894 write_text_atomic(cp, _dump_toml(config))
895 logger.info("✅ Remote %r set to %s", name, url)
896
897
898 def remove_remote(
899 name: str,
900 repo_root: pathlib.Path | None = None,
901 ) -> None:
902 """Remove a named remote and its tracking refs.
903
904 Args:
905 name: Remote name to remove.
906 repo_root: Repository root. Defaults to ``Path.cwd()``.
907
908 Raises:
909 KeyError: If *name* is not a configured remote.
910 """
911 cp = _config_path(repo_root)
912 config = _load_config(cp)
913 remotes = config.get("remotes")
914 if remotes is None or name not in remotes:
915 raise KeyError(name)
916 del remotes[name]
917 config["remotes"] = remotes
918 write_text_atomic(cp, _dump_toml(config))
919 logger.info("✅ Remote %r removed from config", name)
920
921 root = (repo_root or pathlib.Path.cwd()).resolve()
922 refs_dir = root / _MUSE_DIR / "remotes" / name
923 if refs_dir.is_symlink():
924 # Refuse to rmtree a symlink — following a symlink placed by an
925 # attacker could delete files outside the repository tree.
926 logger.warning("⚠️ Skipping rmtree: remotes dir %s is a symlink", refs_dir)
927 elif refs_dir.is_dir():
928 shutil.rmtree(refs_dir)
929 logger.debug("✅ Removed tracking refs dir %s", refs_dir)
930
931
932 def rename_remote(
933 old_name: str,
934 new_name: str,
935 repo_root: pathlib.Path | None = None,
936 ) -> None:
937 """Rename a remote and move its tracking refs.
938
939 Args:
940 old_name: Current remote name.
941 new_name: Desired new remote name.
942 repo_root: Repository root. Defaults to ``Path.cwd()``.
943
944 Raises:
945 KeyError: If *old_name* is not a configured remote.
946 ValueError: If *new_name* is already configured.
947 """
948 cp = _config_path(repo_root)
949 config = _load_config(cp)
950 remotes = config.get("remotes")
951 if remotes is None or old_name not in remotes:
952 raise KeyError(old_name)
953 if new_name in remotes:
954 raise ValueError(new_name)
955 remotes[new_name] = remotes.pop(old_name)
956 config["remotes"] = remotes
957 write_text_atomic(cp, _dump_toml(config))
958 logger.info("✅ Remote %r renamed to %r", old_name, new_name)
959
960 root = (repo_root or pathlib.Path.cwd()).resolve()
961 old_refs_dir = root / _MUSE_DIR / "remotes" / old_name
962 new_refs_dir = root / _MUSE_DIR / "remotes" / new_name
963 if old_refs_dir.is_dir():
964 old_refs_dir.rename(new_refs_dir)
965 logger.debug("✅ Moved tracking refs dir %s → %s", old_refs_dir, new_refs_dir)
966
967
968 def list_remotes(repo_root: pathlib.Path | None = None) -> list[RemoteConfig]:
969 """Return all configured remotes sorted alphabetically by name.
970
971 Args:
972 repo_root: Repository root. Defaults to ``Path.cwd()``.
973
974 Returns:
975 List of ``{"name": str, "url": str}`` dicts.
976 """
977 config = _load_config(_config_path(repo_root))
978 remotes = config.get("remotes")
979 if remotes is None:
980 return []
981 result: list[RemoteConfig] = []
982 for remote_name in sorted(remotes):
983 entry = remotes[remote_name]
984 url = entry.get("url", "")
985 if not url.strip():
986 continue
987 rc = RemoteConfig(name=remote_name, url=url.strip())
988 pack_origin = entry.get("pack_origin", "")
989 if pack_origin.strip():
990 rc["pack_origin"] = pack_origin.strip()
991 result.append(rc)
992 return result
993
994
995 # ---------------------------------------------------------------------------
996 # Remote tracking-head helpers
997 # ---------------------------------------------------------------------------
998
999
1000 def _remote_head_path(
1001 remote_name: str,
1002 branch: str,
1003 repo_root: pathlib.Path | None = None,
1004 ) -> pathlib.Path:
1005 """Return the path to the remote tracking pointer file."""
1006 root = (repo_root or pathlib.Path.cwd()).resolve()
1007 return root / _MUSE_DIR / "remotes" / remote_name / branch
1008
1009
1010 def get_remote_head(
1011 remote_name: str,
1012 branch: str,
1013 repo_root: pathlib.Path | None = None,
1014 ) -> str | None:
1015 """Return the last-known remote commit ID for *remote_name*/*branch*.
1016
1017 Returns ``None`` when the tracking pointer does not exist.
1018
1019 Args:
1020 remote_name: Remote name (e.g. ``"origin"``).
1021 branch: Branch name (e.g. ``"main"``).
1022 repo_root: Repository root. Defaults to ``Path.cwd()``.
1023
1024 Returns:
1025 Commit ID string, or ``None``.
1026 """
1027 pointer = _remote_head_path(remote_name, branch, repo_root)
1028 if not pointer.is_file():
1029 return None
1030 raw = pointer.read_text(encoding="utf-8").strip()
1031 return raw if raw else None
1032
1033
1034 def set_remote_head(
1035 remote_name: str,
1036 branch: str,
1037 commit_id: str,
1038 repo_root: pathlib.Path | None = None,
1039 ) -> None:
1040 """Write the remote tracking pointer for *remote_name*/*branch*.
1041
1042 Args:
1043 remote_name: Remote name (e.g. ``"origin"``).
1044 branch: Branch name.
1045 commit_id: Commit ID to record as the known remote HEAD.
1046 repo_root: Repository root. Defaults to ``Path.cwd()``.
1047 """
1048 pointer = _remote_head_path(remote_name, branch, repo_root)
1049 write_text_atomic(pointer, commit_id)
1050 logger.debug("✅ Remote head %s/%s → %s", remote_name, branch, commit_id[:8])
1051
1052
1053 def delete_remote_head(
1054 remote_name: str,
1055 branch: str,
1056 repo_root: pathlib.Path | None = None,
1057 ) -> bool:
1058 """Remove the local remote-tracking pointer for *remote_name*/*branch*.
1059
1060 Used after ``muse push --delete`` deletes the branch on the server, or when
1061 pruning stale tracking refs with ``muse branch -dr``.
1062
1063 Args:
1064 remote_name: Remote name (e.g. ``"origin"``).
1065 branch: Branch name (e.g. ``"feat/my-thing"``).
1066 repo_root: Repository root. Defaults to ``Path.cwd()``.
1067
1068 Returns:
1069 ``True`` if the pointer file existed and was removed, ``False`` if it
1070 was already absent (idempotent).
1071 """
1072 pointer = _remote_head_path(remote_name, branch, repo_root)
1073 if not pointer.is_file():
1074 return False
1075 pointer.unlink()
1076 # Remove now-empty parent directories (mirrors _cleanup_empty_dirs in branch.py).
1077 remotes_dir = pointer.parent
1078 while remotes_dir.name != remote_name:
1079 try:
1080 remotes_dir.rmdir()
1081 except OSError:
1082 break
1083 remotes_dir = remotes_dir.parent
1084 logger.debug("🗑 Remote tracking ref %s/%s removed", remote_name, branch)
1085 return True
1086
1087
1088 # ---------------------------------------------------------------------------
1089 # Upstream tracking helpers
1090 # ---------------------------------------------------------------------------
1091
1092
1093 def set_upstream(
1094 branch: str,
1095 remote_name: str,
1096 repo_root: pathlib.Path | None = None,
1097 ) -> None:
1098 """Record *remote_name* as the upstream remote for *branch*.
1099
1100 Args:
1101 branch: Local (and remote) branch name.
1102 remote_name: Remote name.
1103 repo_root: Repository root. Defaults to ``Path.cwd()``.
1104 """
1105 cp = _config_path(repo_root)
1106 cp.parent.mkdir(parents=True, exist_ok=True)
1107 config = _load_config(cp)
1108 existing_remotes = config.get("remotes")
1109 remotes: RemotesMap = {}
1110 if existing_remotes:
1111 remotes.update(existing_remotes)
1112 existing_entry = remotes.get(remote_name)
1113 entry: RemoteEntry = {}
1114 if existing_entry is not None:
1115 if "url" in existing_entry:
1116 entry["url"] = existing_entry["url"]
1117 if "branch" in existing_entry:
1118 entry["branch"] = existing_entry["branch"]
1119 entry["branch"] = branch
1120 remotes[remote_name] = entry
1121 config["remotes"] = remotes
1122 write_text_atomic(cp, _dump_toml(config))
1123 logger.info("✅ Upstream for branch %r set to %s/%r", branch, remote_name, branch)
1124
1125
1126 def get_upstream(
1127 branch: str,
1128 repo_root: pathlib.Path | None = None,
1129 ) -> str | None:
1130 """Return the configured upstream remote name for *branch*, or ``None``.
1131
1132 Args:
1133 branch: Local branch name.
1134 repo_root: Repository root. Defaults to ``Path.cwd()``.
1135
1136 Returns:
1137 Remote name string, or ``None``.
1138 """
1139 config = _load_config(_config_path(repo_root))
1140 remotes = config.get("remotes")
1141 if remotes is None:
1142 return None
1143 for rname, entry in remotes.items():
1144 tracked = entry.get("branch", "")
1145 if tracked.strip() == branch:
1146 return rname
1147 return None
1148
1149
1150 # ---------------------------------------------------------------------------
1151 # Global user config — ~/.muse/config.toml (safe_dirs)
1152 # ---------------------------------------------------------------------------
1153
1154 _GLOBAL_MUSE_DIR = pathlib.Path.home() / ".muse"
1155 _GLOBAL_CONFIG_FILE = _GLOBAL_MUSE_DIR / "config.toml"
1156
1157
1158 def _load_global_config() -> dict[str, list[str]]:
1159 """Load ``~/.muse/config.toml`` and return the ``[security]`` section.
1160
1161 Returns a dict with key ``safe_dirs`` mapping to a list of path strings.
1162 Returns ``{"safe_dirs": []}`` when the file is absent or unparseable.
1163 """
1164 if not _GLOBAL_CONFIG_FILE.is_file():
1165 return {"safe_dirs": []}
1166 try:
1167 with _GLOBAL_CONFIG_FILE.open("rb") as fh:
1168 raw: dict[str, object] = tomllib.load(fh)
1169 except Exception as exc: # noqa: BLE001
1170 logger.warning("⚠️ Failed to parse %s: %s", _GLOBAL_CONFIG_FILE, exc)
1171 return {"safe_dirs": []}
1172 security_raw = raw.get("security")
1173 if not isinstance(security_raw, dict):
1174 return {"safe_dirs": []}
1175 dirs_raw = security_raw.get("safe_dirs")
1176 if not isinstance(dirs_raw, list):
1177 return {"safe_dirs": []}
1178 safe: list[str] = [d for d in dirs_raw if isinstance(d, str) and d.strip()]
1179 return {"safe_dirs": safe}
1180
1181
1182 def _save_global_config(safe_dirs: list[str]) -> None:
1183 """Write ``[security] safe_dirs`` to ``~/.muse/config.toml``.
1184
1185 Preserves any other sections that may exist in the file.
1186 """
1187 import os
1188 _GLOBAL_MUSE_DIR.mkdir(parents=True, exist_ok=True)
1189
1190 # Read existing raw content to preserve other sections.
1191 existing_lines: list[str] = []
1192 if _GLOBAL_CONFIG_FILE.is_file():
1193 try:
1194 existing_lines = _GLOBAL_CONFIG_FILE.read_text("utf-8").splitlines()
1195 except Exception: # noqa: BLE001
1196 existing_lines = []
1197
1198 # Strip any existing [security] section from the file.
1199 filtered: list[str] = []
1200 in_security = False
1201 for line in existing_lines:
1202 stripped = line.strip()
1203 if stripped == "[security]":
1204 in_security = True
1205 continue
1206 if in_security and stripped.startswith("["):
1207 in_security = False
1208 if not in_security:
1209 filtered.append(line)
1210
1211 # Remove trailing blank lines before appending the new section.
1212 while filtered and not filtered[-1].strip():
1213 filtered.pop()
1214
1215 # Append the new [security] section.
1216 filtered.append("")
1217 filtered.append("[security]")
1218 if safe_dirs:
1219 items = ", ".join(f'"{_escape(d)}"' for d in safe_dirs)
1220 filtered.append(f"safe_dirs = [{items}]")
1221 else:
1222 filtered.append("safe_dirs = []")
1223 filtered.append("")
1224
1225 content = "\n".join(filtered)
1226 tmp = _GLOBAL_CONFIG_FILE.with_suffix(".toml.tmp")
1227 tmp.write_text(content, encoding="utf-8")
1228 os.replace(tmp, _GLOBAL_CONFIG_FILE)
1229
1230
1231 def get_global_safe_dirs() -> list[str]:
1232 """Return the ``safe_dirs`` list from ``~/.muse/config.toml``.
1233
1234 Returns an empty list when not configured.
1235 """
1236 return _load_global_config().get("safe_dirs", [])
1237
1238
1239 def add_global_safe_dir(path: str) -> None:
1240 """Add *path* to the ``safe_dirs`` list in ``~/.muse/config.toml``.
1241
1242 Normalises the path (``os.path.abspath``) before storing. Idempotent —
1243 adding the same path twice has no effect.
1244
1245 Args:
1246 path: Absolute or relative path to trust.
1247 """
1248 import os
1249 abs_path = os.path.abspath(path)
1250 current = get_global_safe_dirs()
1251 if abs_path not in current:
1252 current.append(abs_path)
1253 _save_global_config(current)
1254 logger.info("✅ Trusted path added: %s", abs_path)
1255
1256
1257 def remove_global_safe_dir(path: str) -> None:
1258 """Remove *path* from the ``safe_dirs`` list in ``~/.muse/config.toml``.
1259
1260 Normalises the path before matching. No-op when the path is not present.
1261
1262 Args:
1263 path: Absolute or relative path to remove from the trust list.
1264 """
1265 import os
1266 abs_path = os.path.abspath(path)
1267 current = get_global_safe_dirs()
1268 updated = [d for d in current if d != abs_path]
1269 _save_global_config(updated)
1270 logger.info("✅ Trusted path removed: %s", abs_path)
File History 2 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 32 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 32 days ago