gabriel / muse public
config.py python
1,083 lines 35.9 KB
Raw
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 151 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
93
94 class LimitsConfig(TypedDict, total=False):
95 """``[limits]`` section in ``.muse/config.toml``.
96
97 All values are optional — defaults are used when absent. Keys map to
98 the ``[limits]`` TOML table::
99
100 [limits]
101 max_walk_commits = 10000 # cap for walk_commits_between / muse log
102 max_ancestors = 50000 # cap for find_merge_base BFS
103 max_graph_commits = 50000 # cap for _collect_all_commits (--graph --all)
104 shard_prefix_length = 2 # object store shard depth: 2 (256 shards)
105 # or 4 (65536 shards) for very large repos
106 """
107
108 max_walk_commits: int
109 max_ancestors: int
110 max_graph_commits: int
111 shard_prefix_length: int
112
113
114 class MuseConfig(TypedDict, total=False):
115 """Structured view of the entire ``.muse/config.toml`` file."""
116
117 user: UserConfig
118 hub: HubConfig
119 remotes: RemotesMap
120 domain: DomainConfig
121 limits: LimitsConfig
122
123
124 class RemoteConfig(TypedDict):
125 """Public-facing remote descriptor returned by :func:`list_remotes`."""
126
127 name: str
128 url: str
129
130
131 # ---------------------------------------------------------------------------
132 # Internal helpers
133 # ---------------------------------------------------------------------------
134
135
136 def _config_path(repo_root: pathlib.Path | None) -> pathlib.Path:
137 """Return the path to .muse/config.toml for the given (or cwd) root."""
138 root = (repo_root or pathlib.Path.cwd()).resolve()
139 return root / _MUSE_DIR / _CONFIG_FILENAME
140
141
142 def _load_config(config_path: pathlib.Path) -> MuseConfig:
143 """Load and parse config.toml; return an empty MuseConfig if absent."""
144 if not config_path.is_file():
145 return {}
146
147 try:
148 with config_path.open("rb") as fh:
149 raw = tomllib.load(fh)
150 except Exception as exc: # noqa: BLE001
151 logger.warning("⚠️ Failed to parse %s: %s", config_path, exc)
152 return {}
153
154 config: MuseConfig = {}
155
156 user_raw = raw.get("user")
157 if isinstance(user_raw, dict):
158 user: UserConfig = {}
159 name_val = user_raw.get("name")
160 if isinstance(name_val, str):
161 user["name"] = name_val
162 email_val = user_raw.get("email")
163 if isinstance(email_val, str):
164 user["email"] = email_val
165 type_val = user_raw.get("type")
166 if isinstance(type_val, str):
167 user["type"] = type_val
168 config["user"] = user
169
170 hub_raw = raw.get("hub")
171 if isinstance(hub_raw, dict):
172 hub: HubConfig = {}
173 url_val = hub_raw.get("url")
174 if isinstance(url_val, str):
175 hub["url"] = url_val
176 config["hub"] = hub
177
178 remotes_raw = raw.get("remotes")
179 if isinstance(remotes_raw, dict):
180 remotes: RemotesMap = {}
181 for name, remote_raw in remotes_raw.items():
182 if isinstance(remote_raw, dict):
183 entry: RemoteEntry = {}
184 rurl = remote_raw.get("url")
185 if isinstance(rurl, str):
186 entry["url"] = rurl
187 branch_val = remote_raw.get("branch")
188 if isinstance(branch_val, str):
189 entry["branch"] = branch_val
190 remotes[name] = entry
191 config["remotes"] = remotes
192
193 domain_raw = raw.get("domain")
194 if isinstance(domain_raw, dict):
195 domain: DomainConfig = {}
196 for key, val in domain_raw.items():
197 if isinstance(val, str):
198 domain[key] = val
199 config["domain"] = domain
200
201 limits_raw = raw.get("limits")
202 if isinstance(limits_raw, dict):
203 limits: LimitsConfig = {}
204 mwc = limits_raw.get("max_walk_commits")
205 if isinstance(mwc, int) and mwc > 0:
206 limits["max_walk_commits"] = mwc
207 ma = limits_raw.get("max_ancestors")
208 if isinstance(ma, int) and ma > 0:
209 limits["max_ancestors"] = ma
210 mgc = limits_raw.get("max_graph_commits")
211 if isinstance(mgc, int) and mgc > 0:
212 limits["max_graph_commits"] = mgc
213 spl = limits_raw.get("shard_prefix_length")
214 if isinstance(spl, int) and spl in (2, 4):
215 limits["shard_prefix_length"] = spl
216 config["limits"] = limits
217
218 return config
219
220
221 def _escape(value: str) -> str:
222 """Escape a TOML basic string value (backslash and double-quote only).
223
224 TOML basic strings allow control characters escaped as ``\\n``, ``\\t``,
225 etc., but we store only printable content — control characters in values
226 are also stripped here so the resulting TOML file remains parseable.
227 """
228 return (
229 value.replace("\\", "\\\\")
230 .replace('"', '\\"')
231 .replace("\n", "\\n")
232 .replace("\r", "\\r")
233 .replace("\0", "")
234 )
235
236
237 # Characters that are structurally significant in unquoted TOML keys and
238 # table headers. Any of these in a key name would allow injection of
239 # arbitrary TOML sections or key-value pairs.
240 _TOML_KEY_UNSAFE: frozenset[str] = frozenset('\n\r\0][="')
241
242
243 def _validate_toml_key(key: str, context: str = "key") -> None:
244 """Raise ``ValueError`` if *key* contains TOML-structurally unsafe characters.
245
246 Prevents injection attacks where a crafted key like ``x]\\n[evil`` would
247 break the TOML section structure and allow writing arbitrary sections.
248
249 Args:
250 key: Key string to validate.
251 context: Human-readable label used in the error message (e.g. ``"domain key"``).
252
253 Raises:
254 ValueError: If any character in *key* is in ``_TOML_KEY_UNSAFE``.
255 """
256 bad = _TOML_KEY_UNSAFE & set(key)
257 if bad:
258 chars = ", ".join(sorted(repr(c) for c in bad))
259 raise ValueError(
260 f"Config {context} {key!r} contains characters not allowed in TOML keys: {chars}"
261 )
262
263
264 def _dump_toml(config: MuseConfig) -> str:
265 """Serialise a MuseConfig to TOML text.
266
267 Section order: ``[user]``, ``[hub]``, ``[remotes.*]``, ``[domain]``, ``[limits]``.
268
269 All key names are validated against ``_TOML_KEY_UNSAFE`` before being
270 written, preventing TOML injection via crafted domain keys or remote names.
271 """
272 lines: list[str] = []
273
274 user = config.get("user")
275 if user:
276 lines.append("[user]")
277 name = user.get("name", "")
278 if name:
279 lines.append(f'name = "{_escape(name)}"')
280 email = user.get("email", "")
281 if email:
282 lines.append(f'email = "{_escape(email)}"')
283 utype = user.get("type", "")
284 if utype:
285 lines.append(f'type = "{_escape(utype)}"')
286 lines.append("")
287
288 hub = config.get("hub")
289 if hub:
290 lines.append("[hub]")
291 url = hub.get("url", "")
292 if url:
293 lines.append(f'url = "{_escape(url)}"')
294 lines.append("")
295
296 remotes = config.get("remotes") or {}
297 for remote_name in sorted(remotes):
298 # Remote names come from _load_config which parses TOML, so they are
299 # safe at read time. Validate defensively before writing.
300 _validate_toml_key(remote_name, "remote name")
301 entry = remotes[remote_name]
302 lines.append(f"[remotes.{remote_name}]")
303 rurl = entry.get("url", "")
304 if rurl:
305 lines.append(f'url = "{_escape(rurl)}"')
306 branch = entry.get("branch", "")
307 if branch:
308 lines.append(f'branch = "{_escape(branch)}"')
309 lines.append("")
310
311 domain = config.get("domain") or {}
312 if domain:
313 lines.append("[domain]")
314 for key, val in sorted(domain.items()):
315 _validate_toml_key(key, "domain key")
316 lines.append(f'{key} = "{_escape(val)}"')
317 lines.append("")
318
319 limits = config.get("limits") or {}
320 if limits:
321 lines.append("[limits]")
322 mwc = limits.get("max_walk_commits")
323 if mwc is not None:
324 lines.append(f"max_walk_commits = {mwc}")
325 ma = limits.get("max_ancestors")
326 if ma is not None:
327 lines.append(f"max_ancestors = {ma}")
328 mgc = limits.get("max_graph_commits")
329 if mgc is not None:
330 lines.append(f"max_graph_commits = {mgc}")
331 spl = limits.get("shard_prefix_length")
332 if spl is not None:
333 lines.append(f"shard_prefix_length = {spl}")
334 lines.append("")
335
336 return "\n".join(lines)
337
338
339 # ---------------------------------------------------------------------------
340 # Auth token resolution (via identity store)
341 # ---------------------------------------------------------------------------
342
343
344 def get_signing_identity(
345 repo_root: pathlib.Path | None = None,
346 remote_url: str | None = None,
347 agent_id: str | None = None,
348 ) -> "object | None":
349 """Return a :class:`~muse.core.transport.SigningIdentity` for a hub, or ``None``.
350
351 Resolution order:
352 1. ``MUSE_AGENT_KEY`` environment variable — PEM private key bytes. When
353 set the key is used directly, bypassing the identity store entirely.
354 The handle is taken from ``MUSE_AGENT_HANDLE`` (defaults to *agent_id*
355 if that is also set, otherwise ``"agent"``). This is the injection
356 mechanism for agent subprocesses spawned by agentception.
357 2. Agent-specific entry in ``~/.muse/identity.toml`` keyed by
358 ``"hostname#agent_id"`` — when *agent_id* is provided.
359 3. Human entry in ``~/.muse/identity.toml`` keyed by bare hostname.
360 4. Hub URL from ``[hub] url`` in ``.muse/config.toml`` (fallback lookup
361 URL when *remote_url* is not supplied).
362
363 The private key is **never** logged.
364
365 Args:
366 repo_root: Repository root. Defaults to ``Path.cwd()``.
367 remote_url: URL of the specific remote being contacted.
368 agent_id: Agent handle, e.g. ``"agentception-abc123"``. Used to
369 try an agent-specific key before falling back to the
370 human key.
371
372 Returns:
373 :class:`~muse.core.transport.SigningIdentity` or ``None``.
374 """
375 import os as _os
376 from muse.core.identity import resolve_signing_identity # avoid circular import
377 from muse.core.transport import SigningIdentity
378
379 # 1. MUSE_AGENT_KEY env var — injected by agentception at spawn time.
380 agent_pem = _os.environ.get("MUSE_AGENT_KEY", "").strip()
381 if agent_pem:
382 from muse.core.keypair import load_private_key_from_pem
383 private_key = load_private_key_from_pem(agent_pem.encode())
384 if private_key is not None:
385 handle = (
386 _os.environ.get("MUSE_AGENT_HANDLE", "").strip()
387 or agent_id
388 or "agent"
389 )
390 logger.debug("✅ Signing identity from MUSE_AGENT_KEY (handle=%s)", handle)
391 return SigningIdentity(handle=handle, private_key=private_key)
392 logger.warning("⚠️ MUSE_AGENT_KEY is set but could not be decoded — ignoring")
393
394 # 2 & 3. Identity store lookup (agent key → human key fallback).
395 lookup_url: str | None = remote_url or get_hub_url(repo_root)
396 if lookup_url is None:
397 logger.debug("⚠️ No hub configured — skipping signing identity lookup")
398 return None
399
400 result = resolve_signing_identity(lookup_url, agent_id=agent_id)
401 if result is None:
402 logger.debug(
403 "⚠️ No signing identity for hub %s — run `muse auth keygen && muse auth register`",
404 lookup_url,
405 )
406 return None
407
408 handle, private_key = result
409 logger.debug("✅ Signing identity resolved for hub %s (handle=%s)", lookup_url, handle)
410 return SigningIdentity(handle=handle, private_key=private_key)
411
412
413
414
415 # ---------------------------------------------------------------------------
416 # Hub helpers
417 # ---------------------------------------------------------------------------
418
419
420 def get_hub_url(repo_root: pathlib.Path | None = None) -> str | None:
421 """Return the hub URL from ``[hub] url``, or ``None`` if not configured.
422
423 Args:
424 repo_root: Repository root. Defaults to ``Path.cwd()``.
425
426 Returns:
427 URL string, or ``None``.
428 """
429 config = _load_config(_config_path(repo_root))
430 hub = config.get("hub")
431 if hub is None:
432 return None
433 url = hub.get("url", "")
434 return url.strip() if url.strip() else None
435
436
437 def set_hub_url(url: str, repo_root: pathlib.Path | None = None) -> None:
438 """Write ``[hub] url`` to ``.muse/config.toml``.
439
440 Preserves all other sections. Creates the config file if absent.
441 Rejects ``http://`` URLs — Muse never contacts a hub over cleartext HTTP.
442
443 Args:
444 url: Hub URL (must be ``https://``).
445 repo_root: Repository root. Defaults to ``Path.cwd()``.
446
447 Raises:
448 ValueError: If *url* does not use the ``https://`` scheme.
449 """
450 _is_loopback = url.startswith("http://localhost") or url.startswith("http://127.0.0.1") or url.startswith("http://[::1]")
451 if not url.startswith("https://") and not _is_loopback:
452 raise ValueError(
453 f"Hub URL must use HTTPS. Got: {url!r}\n"
454 "Muse never connects to a hub over cleartext HTTP.\n"
455 "(Exception: http://localhost and http://127.0.0.1 are allowed for local development.)"
456 )
457 cp = _config_path(repo_root)
458 cp.parent.mkdir(parents=True, exist_ok=True)
459 config = _load_config(cp)
460 config["hub"] = HubConfig(url=url)
461 write_text_atomic(cp, _dump_toml(config))
462 logger.info("✅ Hub URL set to %s", url)
463
464
465 def clear_hub_url(repo_root: pathlib.Path | None = None) -> None:
466 """Remove the ``[hub]`` section from ``.muse/config.toml``.
467
468 Args:
469 repo_root: Repository root. Defaults to ``Path.cwd()``.
470 """
471 cp = _config_path(repo_root)
472 config = _load_config(cp)
473 if "hub" in config:
474 del config["hub"]
475 write_text_atomic(cp, _dump_toml(config))
476 logger.info("✅ Hub disconnected")
477
478
479 # ---------------------------------------------------------------------------
480 # User config helpers
481 # ---------------------------------------------------------------------------
482
483
484
485 def set_user_field(key: str, value: str, repo_root: pathlib.Path | None = None) -> None:
486 """Set a single ``[user]`` field by name.
487
488 Allowed keys: ``name``, ``email``, ``type``.
489
490 Args:
491 key: Field name within ``[user]``.
492 value: New value.
493 repo_root: Repository root. Defaults to ``Path.cwd()``.
494
495 Raises:
496 ValueError: If *key* is not a recognised user config field.
497 """
498 if key not in {"name", "email", "type"}:
499 raise ValueError(f"Unknown [user] config key: {key!r}. Valid keys: name, email, type")
500 cp = _config_path(repo_root)
501 cp.parent.mkdir(parents=True, exist_ok=True)
502 config = _load_config(cp)
503 user: UserConfig = config.get("user") or {}
504 if key == "name":
505 user["name"] = value
506 elif key == "email":
507 user["email"] = value
508 elif key == "type":
509 user["type"] = value
510 config["user"] = user
511 write_text_atomic(cp, _dump_toml(config))
512 logger.info("✅ user.%s = %r", key, value)
513
514
515 # ---------------------------------------------------------------------------
516 # Generic dotted-key helpers
517 # ---------------------------------------------------------------------------
518
519 _BlockedNS = dict[str, str]
520 _BLOCKED_NAMESPACES: _BlockedNS = {
521 "auth": "Use `muse auth keygen` and `muse auth register` to manage credentials.",
522 "remotes": "Use `muse remote add/remove/rename` to manage remotes.",
523 }
524
525 _SETTABLE_NAMESPACES = {"user", "hub", "domain", "limits"}
526
527 # Default cap values — used when the [limits] section is absent or the key
528 # is not set. These are the same values that were previously hardcoded inside
529 # the individual functions.
530 _DEFAULT_MAX_WALK_COMMITS: int = 10_000
531 _DEFAULT_MAX_ANCESTORS: int = 50_000
532 _DEFAULT_MAX_GRAPH_COMMITS: int = 50_000
533 _DEFAULT_SHARD_PREFIX_LENGTH: int = 2
534
535
536 def get_limit(key: str, repo_root: pathlib.Path | None = None) -> int:
537 """Return a ``[limits]`` integer cap from config, or its default.
538
539 Args:
540 key: Limit key — one of ``max_walk_commits``, ``max_ancestors``,
541 ``max_graph_commits``.
542 repo_root: Repository root; ``None`` falls back to ``Path.cwd()``.
543
544 Returns:
545 Configured integer value, or the built-in default if not set.
546 """
547 defaults: DefaultsMap = {
548 "max_walk_commits": _DEFAULT_MAX_WALK_COMMITS,
549 "max_ancestors": _DEFAULT_MAX_ANCESTORS,
550 "max_graph_commits": _DEFAULT_MAX_GRAPH_COMMITS,
551 "shard_prefix_length": _DEFAULT_SHARD_PREFIX_LENGTH,
552 }
553 default = defaults.get(key, 10_000)
554 config = _load_config(_config_path(repo_root))
555 limits = config.get("limits") or {}
556 # Explicit key dispatch keeps mypy happy on TypedDict literal-required keys.
557 if key == "max_walk_commits":
558 val: int | None = limits.get("max_walk_commits")
559 elif key == "max_ancestors":
560 val = limits.get("max_ancestors")
561 elif key == "max_graph_commits":
562 val = limits.get("max_graph_commits")
563 elif key == "shard_prefix_length":
564 val = limits.get("shard_prefix_length")
565 else:
566 val = None
567 if isinstance(val, int) and val > 0:
568 return val
569 return default
570
571
572 def get_config_value(key: str, repo_root: pathlib.Path | None = None) -> str | None:
573 """Get a config value by dotted key (e.g. ``user.name``, ``hub.url``).
574
575 Returns ``None`` when the key is not set or the namespace is unknown.
576
577 Args:
578 key: Dotted key in ``<namespace>.<subkey>`` form.
579 repo_root: Repository root. Defaults to ``Path.cwd()``.
580
581 Returns:
582 String value, or ``None``.
583 """
584 parts = key.split(".", 1)
585 if len(parts) != 2:
586 return None
587 namespace, subkey = parts
588 config = _load_config(_config_path(repo_root))
589
590 if namespace == "user":
591 user = config.get("user") or {}
592 if subkey == "name":
593 return user.get("name")
594 if subkey == "email":
595 return user.get("email")
596 if subkey == "type":
597 return user.get("type")
598 return None
599
600 if namespace == "hub":
601 hub = config.get("hub") or {}
602 if subkey == "url":
603 return hub.get("url")
604 return None
605
606 if namespace == "domain":
607 domain = config.get("domain") or {}
608 return domain.get(subkey)
609
610 if namespace == "limits":
611 limits = config.get("limits") or {}
612 if subkey == "max_walk_commits":
613 v = limits.get("max_walk_commits")
614 return str(v) if isinstance(v, int) else None
615 if subkey == "max_ancestors":
616 v = limits.get("max_ancestors")
617 return str(v) if isinstance(v, int) else None
618 if subkey == "max_graph_commits":
619 v = limits.get("max_graph_commits")
620 return str(v) if isinstance(v, int) else None
621 if subkey == "shard_prefix_length":
622 v = limits.get("shard_prefix_length")
623 return str(v) if isinstance(v, int) else None
624 return None
625
626 return None
627
628
629 def set_config_value(key: str, value: str, repo_root: pathlib.Path | None = None) -> None:
630 """Set a config value by dotted key (e.g. ``user.name``, ``domain.ticks_per_beat``).
631
632 Args:
633 key: Dotted key in ``<namespace>.<subkey>`` form.
634 value: New string value.
635 repo_root: Repository root. Defaults to ``Path.cwd()``.
636
637 Raises:
638 ValueError: If the namespace is blocked, unknown, or the subkey is invalid.
639 """
640 parts = key.split(".", 1)
641 if len(parts) != 2:
642 raise ValueError(f"Key must be in 'namespace.subkey' form, got: {key!r}")
643 namespace, subkey = parts
644
645 if namespace in _BLOCKED_NAMESPACES:
646 raise ValueError(_BLOCKED_NAMESPACES[namespace])
647
648 if namespace not in _SETTABLE_NAMESPACES:
649 raise ValueError(
650 f"Unknown config namespace {namespace!r}. "
651 f"Settable namespaces: {', '.join(sorted(_SETTABLE_NAMESPACES))}"
652 )
653
654 cp = _config_path(repo_root)
655 cp.parent.mkdir(parents=True, exist_ok=True)
656 config = _load_config(cp)
657
658 if namespace == "user":
659 set_user_field(subkey, value, repo_root)
660 return
661
662 if namespace == "hub":
663 if subkey != "url":
664 raise ValueError(f"Unknown [hub] config key: {subkey!r}. Valid keys: url")
665 # Route through set_hub_url — it enforces the HTTPS requirement.
666 set_hub_url(value, repo_root)
667 return
668
669 if namespace == "limits":
670 _LIMITS_KEYS = frozenset({
671 "max_walk_commits", "max_ancestors", "max_graph_commits", "shard_prefix_length",
672 })
673 if subkey not in _LIMITS_KEYS:
674 raise ValueError(
675 f"Unknown [limits] config key: {subkey!r}. "
676 f"Valid keys: {', '.join(sorted(_LIMITS_KEYS))}"
677 )
678 try:
679 int_value = int(value)
680 except ValueError as exc:
681 raise ValueError(
682 f"[limits] {subkey} must be an integer, got: {value!r}"
683 ) from exc
684 if int_value <= 0:
685 raise ValueError(f"[limits] {subkey} must be a positive integer, got: {int_value}")
686 if subkey == "shard_prefix_length" and int_value not in (2, 4):
687 raise ValueError("shard_prefix_length must be 2 or 4")
688 limits_section: LimitsConfig = config.get("limits") or {}
689 if subkey == "max_walk_commits":
690 limits_section["max_walk_commits"] = int_value
691 elif subkey == "max_ancestors":
692 limits_section["max_ancestors"] = int_value
693 elif subkey == "max_graph_commits":
694 limits_section["max_graph_commits"] = int_value
695 elif subkey == "shard_prefix_length":
696 limits_section["shard_prefix_length"] = int_value
697 config["limits"] = limits_section
698 write_text_atomic(cp, _dump_toml(config))
699 logger.info("✅ limits.%s = %d", subkey, int_value)
700 return
701
702 # namespace == "domain"
703 _validate_toml_key(subkey, "domain key")
704 domain: DomainConfig = config.get("domain") or {}
705 domain[subkey] = value
706 config["domain"] = domain
707 write_text_atomic(cp, _dump_toml(config))
708 logger.info("✅ domain.%s = %r", subkey, value)
709
710
711 def config_as_dict(repo_root: pathlib.Path | None = None) -> ConfigTree:
712 """Return the full config as a plain ``dict[str, dict[str, str]]`` for JSON output.
713
714 Credentials are never included — the hub section only contains the URL.
715
716 Args:
717 repo_root: Repository root. Defaults to ``Path.cwd()``.
718
719 Returns:
720 Nested dict suitable for ``json.dumps``.
721 """
722 config = _load_config(_config_path(repo_root))
723 result: ConfigTree = {}
724
725 user = config.get("user")
726 if user:
727 user_dict: ConfigSection = {}
728 uname = user.get("name")
729 if uname:
730 user_dict["name"] = uname
731 uemail = user.get("email")
732 if uemail:
733 user_dict["email"] = uemail
734 utype = user.get("type")
735 if utype:
736 user_dict["type"] = utype
737 if user_dict:
738 result["user"] = user_dict
739
740 hub = config.get("hub")
741 if hub:
742 hub_url = hub.get("url", "")
743 if hub_url:
744 result["hub"] = {"url": hub_url}
745
746 remotes = config.get("remotes") or {}
747 if remotes:
748 remotes_dict: ConfigSection = {}
749 for rname, entry in sorted(remotes.items()):
750 url = entry.get("url", "")
751 if url:
752 remotes_dict[rname] = url
753 if remotes_dict:
754 result["remotes"] = remotes_dict
755
756 domain = config.get("domain") or {}
757 if domain:
758 result["domain"] = dict(sorted(domain.items()))
759
760 limits = config.get("limits") or {}
761 if limits:
762 limits_dict: ConfigSection = {}
763 for lk in ("max_walk_commits", "max_ancestors", "max_graph_commits", "shard_prefix_length"):
764 lv = limits.get(lk)
765 if lv is not None:
766 limits_dict[lk] = str(lv)
767 if limits_dict:
768 result["limits"] = limits_dict
769
770 return result
771
772
773 def config_path_for_editor(repo_root: pathlib.Path | None = None) -> pathlib.Path:
774 """Return the config path for the ``config edit`` command."""
775 return _config_path(repo_root)
776
777
778 # ---------------------------------------------------------------------------
779 # Remote helpers
780 # ---------------------------------------------------------------------------
781
782
783 def get_remote(name: str, repo_root: pathlib.Path | None = None) -> str | None:
784 """Return the URL for remote *name*, or ``None`` when not configured.
785
786 Args:
787 name: Remote name (e.g. ``"origin"``).
788 repo_root: Repository root. Defaults to ``Path.cwd()``.
789
790 Returns:
791 URL string, or ``None``.
792 """
793 config = _load_config(_config_path(repo_root))
794 remotes = config.get("remotes")
795 if remotes is None:
796 return None
797 entry = remotes.get(name)
798 if entry is None:
799 return None
800 url = entry.get("url", "")
801 return url.strip() if url.strip() else None
802
803
804 def set_remote(
805 name: str,
806 url: str,
807 repo_root: pathlib.Path | None = None,
808 ) -> None:
809 """Write ``[remotes.<name>] url`` to ``.muse/config.toml``.
810
811 Preserves all other sections. Creates the file if absent.
812
813 Args:
814 name: Remote name (e.g. ``"origin"``).
815 url: Remote URL.
816 repo_root: Repository root. Defaults to ``Path.cwd()``.
817 """
818 cp = _config_path(repo_root)
819 cp.parent.mkdir(parents=True, exist_ok=True)
820 config = _load_config(cp)
821 existing_remotes = config.get("remotes")
822 remotes: RemotesMap = {}
823 if existing_remotes:
824 remotes.update(existing_remotes)
825 existing_entry = remotes.get(name)
826 entry: RemoteEntry = {}
827 if existing_entry is not None:
828 if "url" in existing_entry:
829 entry["url"] = existing_entry["url"]
830 if "branch" in existing_entry:
831 entry["branch"] = existing_entry["branch"]
832 entry["url"] = url
833 remotes[name] = entry
834 config["remotes"] = remotes
835 write_text_atomic(cp, _dump_toml(config))
836 logger.info("✅ Remote %r set to %s", name, url)
837
838
839 def remove_remote(
840 name: str,
841 repo_root: pathlib.Path | None = None,
842 ) -> None:
843 """Remove a named remote and its tracking refs.
844
845 Args:
846 name: Remote name to remove.
847 repo_root: Repository root. Defaults to ``Path.cwd()``.
848
849 Raises:
850 KeyError: If *name* is not a configured remote.
851 """
852 cp = _config_path(repo_root)
853 config = _load_config(cp)
854 remotes = config.get("remotes")
855 if remotes is None or name not in remotes:
856 raise KeyError(name)
857 del remotes[name]
858 config["remotes"] = remotes
859 write_text_atomic(cp, _dump_toml(config))
860 logger.info("✅ Remote %r removed from config", name)
861
862 root = (repo_root or pathlib.Path.cwd()).resolve()
863 refs_dir = root / _MUSE_DIR / "remotes" / name
864 if refs_dir.is_symlink():
865 # Refuse to rmtree a symlink — following a symlink placed by an
866 # attacker could delete files outside the repository tree.
867 logger.warning("⚠️ Skipping rmtree: remotes dir %s is a symlink", refs_dir)
868 elif refs_dir.is_dir():
869 shutil.rmtree(refs_dir)
870 logger.debug("✅ Removed tracking refs dir %s", refs_dir)
871
872
873 def rename_remote(
874 old_name: str,
875 new_name: str,
876 repo_root: pathlib.Path | None = None,
877 ) -> None:
878 """Rename a remote and move its tracking refs.
879
880 Args:
881 old_name: Current remote name.
882 new_name: Desired new remote name.
883 repo_root: Repository root. Defaults to ``Path.cwd()``.
884
885 Raises:
886 KeyError: If *old_name* is not a configured remote.
887 ValueError: If *new_name* is already configured.
888 """
889 cp = _config_path(repo_root)
890 config = _load_config(cp)
891 remotes = config.get("remotes")
892 if remotes is None or old_name not in remotes:
893 raise KeyError(old_name)
894 if new_name in remotes:
895 raise ValueError(new_name)
896 remotes[new_name] = remotes.pop(old_name)
897 config["remotes"] = remotes
898 write_text_atomic(cp, _dump_toml(config))
899 logger.info("✅ Remote %r renamed to %r", old_name, new_name)
900
901 root = (repo_root or pathlib.Path.cwd()).resolve()
902 old_refs_dir = root / _MUSE_DIR / "remotes" / old_name
903 new_refs_dir = root / _MUSE_DIR / "remotes" / new_name
904 if old_refs_dir.is_dir():
905 old_refs_dir.rename(new_refs_dir)
906 logger.debug("✅ Moved tracking refs dir %s → %s", old_refs_dir, new_refs_dir)
907
908
909 def list_remotes(repo_root: pathlib.Path | None = None) -> list[RemoteConfig]:
910 """Return all configured remotes sorted alphabetically by name.
911
912 Args:
913 repo_root: Repository root. Defaults to ``Path.cwd()``.
914
915 Returns:
916 List of ``{"name": str, "url": str}`` dicts.
917 """
918 config = _load_config(_config_path(repo_root))
919 remotes = config.get("remotes")
920 if remotes is None:
921 return []
922 result: list[RemoteConfig] = []
923 for remote_name in sorted(remotes):
924 entry = remotes[remote_name]
925 url = entry.get("url", "")
926 if url.strip():
927 result.append(RemoteConfig(name=remote_name, url=url.strip()))
928 return result
929
930
931 # ---------------------------------------------------------------------------
932 # Remote tracking-head helpers
933 # ---------------------------------------------------------------------------
934
935
936 def _remote_head_path(
937 remote_name: str,
938 branch: str,
939 repo_root: pathlib.Path | None = None,
940 ) -> pathlib.Path:
941 """Return the path to the remote tracking pointer file."""
942 root = (repo_root or pathlib.Path.cwd()).resolve()
943 return root / _MUSE_DIR / "remotes" / remote_name / branch
944
945
946 def get_remote_head(
947 remote_name: str,
948 branch: str,
949 repo_root: pathlib.Path | None = None,
950 ) -> str | None:
951 """Return the last-known remote commit ID for *remote_name*/*branch*.
952
953 Returns ``None`` when the tracking pointer does not exist.
954
955 Args:
956 remote_name: Remote name (e.g. ``"origin"``).
957 branch: Branch name (e.g. ``"main"``).
958 repo_root: Repository root. Defaults to ``Path.cwd()``.
959
960 Returns:
961 Commit ID string, or ``None``.
962 """
963 pointer = _remote_head_path(remote_name, branch, repo_root)
964 if not pointer.is_file():
965 return None
966 raw = pointer.read_text(encoding="utf-8").strip()
967 return raw if raw else None
968
969
970 def set_remote_head(
971 remote_name: str,
972 branch: str,
973 commit_id: str,
974 repo_root: pathlib.Path | None = None,
975 ) -> None:
976 """Write the remote tracking pointer for *remote_name*/*branch*.
977
978 Args:
979 remote_name: Remote name (e.g. ``"origin"``).
980 branch: Branch name.
981 commit_id: Commit ID to record as the known remote HEAD.
982 repo_root: Repository root. Defaults to ``Path.cwd()``.
983 """
984 pointer = _remote_head_path(remote_name, branch, repo_root)
985 write_text_atomic(pointer, commit_id)
986 logger.debug("✅ Remote head %s/%s → %s", remote_name, branch, commit_id[:8])
987
988
989 def delete_remote_head(
990 remote_name: str,
991 branch: str,
992 repo_root: pathlib.Path | None = None,
993 ) -> bool:
994 """Remove the local remote-tracking pointer for *remote_name*/*branch*.
995
996 Used after ``muse push --delete`` deletes the branch on the server, or when
997 pruning stale tracking refs with ``muse branch -dr``.
998
999 Args:
1000 remote_name: Remote name (e.g. ``"origin"``).
1001 branch: Branch name (e.g. ``"feat/my-thing"``).
1002 repo_root: Repository root. Defaults to ``Path.cwd()``.
1003
1004 Returns:
1005 ``True`` if the pointer file existed and was removed, ``False`` if it
1006 was already absent (idempotent).
1007 """
1008 pointer = _remote_head_path(remote_name, branch, repo_root)
1009 if not pointer.is_file():
1010 return False
1011 pointer.unlink()
1012 # Remove now-empty parent directories (mirrors _cleanup_empty_dirs in branch.py).
1013 remotes_dir = pointer.parent
1014 while remotes_dir.name != remote_name:
1015 try:
1016 remotes_dir.rmdir()
1017 except OSError:
1018 break
1019 remotes_dir = remotes_dir.parent
1020 logger.debug("🗑 Remote tracking ref %s/%s removed", remote_name, branch)
1021 return True
1022
1023
1024 # ---------------------------------------------------------------------------
1025 # Upstream tracking helpers
1026 # ---------------------------------------------------------------------------
1027
1028
1029 def set_upstream(
1030 branch: str,
1031 remote_name: str,
1032 repo_root: pathlib.Path | None = None,
1033 ) -> None:
1034 """Record *remote_name* as the upstream remote for *branch*.
1035
1036 Args:
1037 branch: Local (and remote) branch name.
1038 remote_name: Remote name.
1039 repo_root: Repository root. Defaults to ``Path.cwd()``.
1040 """
1041 cp = _config_path(repo_root)
1042 cp.parent.mkdir(parents=True, exist_ok=True)
1043 config = _load_config(cp)
1044 existing_remotes = config.get("remotes")
1045 remotes: RemotesMap = {}
1046 if existing_remotes:
1047 remotes.update(existing_remotes)
1048 existing_entry = remotes.get(remote_name)
1049 entry: RemoteEntry = {}
1050 if existing_entry is not None:
1051 if "url" in existing_entry:
1052 entry["url"] = existing_entry["url"]
1053 if "branch" in existing_entry:
1054 entry["branch"] = existing_entry["branch"]
1055 entry["branch"] = branch
1056 remotes[remote_name] = entry
1057 config["remotes"] = remotes
1058 write_text_atomic(cp, _dump_toml(config))
1059 logger.info("✅ Upstream for branch %r set to %s/%r", branch, remote_name, branch)
1060
1061
1062 def get_upstream(
1063 branch: str,
1064 repo_root: pathlib.Path | None = None,
1065 ) -> str | None:
1066 """Return the configured upstream remote name for *branch*, or ``None``.
1067
1068 Args:
1069 branch: Local branch name.
1070 repo_root: Repository root. Defaults to ``Path.cwd()``.
1071
1072 Returns:
1073 Remote name string, or ``None``.
1074 """
1075 config = _load_config(_config_path(repo_root))
1076 remotes = config.get("remotes")
1077 if remotes is None:
1078 return None
1079 for rname, entry in remotes.items():
1080 tracked = entry.get("branch", "")
1081 if tracked.strip() == branch:
1082 return rname
1083 return None
File History 1 commit
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 151 days ago