gabriel / muse public
config.py python
1,524 lines 54.9 KB
Raw
sha256:94f494a8f59e4b708ebb89e304209737ce34324a33aae83840a6f6b06d7b8d9d docs: add domain-extensibility.md — the two-axis breadth/de… Sonnet 5 3 days ago
1 """Muse CLI configuration helpers.
2
3 Reads and writes ``.muse/config.toml`` — the per-repository configuration
4 file. Credentials and user identity 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 [hub]
12 url = "https://musehub.ai" # MuseHub fabric endpoint for this repo
13
14 [remotes.origin]
15 url = "https://hub.muse.io/repos/my-repo"
16 branch = "main"
17
18 [domain]
19 # Domain-specific key/value pairs; read by the active domain plugin.
20 # ticks_per_beat = "480"
21
22 Settable via ``muse config set``
23 ---------------------------------
24 - ``hub.url`` (alias: ``muse hub connect <url>``)
25 - ``domain.*``
26
27 Not settable via ``muse config set``
28 --------------------------------------
29 - ``user.*`` — use ``muse auth register`` / ``muse auth whoami``
30 - ``remotes.*`` — use ``muse remote add/remove``
31 - credentials — use ``muse auth register``
32
33 Token resolution
34 ----------------
35 :func:`get_signing_identity` reads the hub URL from this file, then resolves the
36 signing identity from ``~/.muse/identity.toml`` via
37 :func:`muse.core.identity.resolve_token`. The token is **never** logged.
38 """
39
40 import fnmatch
41 import logging
42 import pathlib
43 import shutil
44 import subprocess
45 import tomllib
46 from typing import TypedDict
47
48 from muse.core.types import short_id
49 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
50 from muse.core.refs import read_ref
51 from muse.core.io import write_text_atomic
52
53 logger = logging.getLogger(__name__)
54
55 type RemotesMap = dict[str, RemoteEntry] # remote_name → remote entry
56 type DomainConfig = dict[str, str] # domain key → value
57 type ConfigSection = dict[str, str] # generic flattened section dict
58 type ConfigTree = dict[str, ConfigSection] # section → key → value (for JSON output)
59 type DefaultsMap = dict[str, int] # config key → default int value
60 type _SecurityConfig = dict[str, list[str]] # security section from global config
61
62 # ---------------------------------------------------------------------------
63 # Named configuration types
64 # ---------------------------------------------------------------------------
65
66 class HubConfig(TypedDict, total=False):
67 """``[hub]`` section in ``.muse/config.toml``."""
68
69 url: str
70
71 class RemoteEntry(TypedDict, total=False):
72 """``[remotes.<name>]`` section in ``.muse/config.toml``."""
73
74 url: str
75 branch: str
76 promisor: bool # when False, this remote is not used as a promisor for missing objects
77
78 class LimitsConfig(TypedDict, total=False):
79 """``[limits]`` section in ``.muse/config.toml``.
80
81 All values are optional — defaults are used when absent. Keys map to
82 the ``[limits]`` TOML table::
83
84 [limits]
85 max_walk_commits = 10000 # cap for walk_commits_between / muse log
86 max_ancestors = 50000 # cap for find_merge_base BFS
87 max_graph_commits = 50000 # cap for _collect_all_commits (--graph --all)
88 shard_prefix_length = 2 # object store shard depth: 2 (256 shards)
89 # or 4 (65536 shards) for very large repos
90 """
91
92 max_walk_commits: int
93 max_ancestors: int
94 max_graph_commits: int
95 shard_prefix_length: int
96
97 class CommitConfig(TypedDict, total=False):
98 """``[commit]`` section in ``.muse/config.toml``."""
99
100 sign: bool
101
102
103 class SymlogConfig(TypedDict, total=False):
104 """``[symlog]`` section in ``.muse/config.toml``.
105
106 Controls default TTL for per-symbol journal expiry::
107
108 [symlog]
109 expire_days = 90 # entries older than this are pruned by muse gc / muse symlog expire
110 """
111
112 expire_days: int
113
114
115 class PushConfig(TypedDict, total=False):
116 """``[push]`` section in ``.muse/config.toml``."""
117
118 tags: bool # when true, version tags are pushed on every muse push
119
120
121 class BranchMeta(TypedDict, total=False):
122 """Per-branch metadata stored under ``[branch."<name>"]`` in config.toml.
123
124 Fields written by ``muse branch --intent / --resumable``::
125
126 [branch."feat/my-thing"]
127 intent = "refactor auth layer"
128 resumable = true
129
130 Fields written by ``muse push`` upstream tracking (preserved on read/write)::
131
132 remote = "origin"
133 merge = "refs/heads/feat/my-thing"
134 """
135
136 intent: str # short description of what this branch is for
137 resumable: bool # true when this branch is a resumable agent checkpoint
138 remote: str # upstream remote name (e.g. "origin")
139 merge: str # upstream merge ref (e.g. "refs/heads/main")
140
141 class MuseConfig(TypedDict, total=False):
142 """Structured view of the entire ``.muse/config.toml`` file."""
143
144 hub: HubConfig
145 remotes: RemotesMap
146 symlog: SymlogConfig
147 domain: DomainConfig
148 limits: LimitsConfig
149 commit: CommitConfig
150 push: PushConfig
151 branch: "dict[str, BranchMeta]" # branch_name → per-branch metadata
152 protected_branches: "list[str]" # fnmatch patterns from [protected_branches]
153
154 class RemoteConfig(TypedDict, total=False):
155 """Public-facing remote descriptor returned by :func:`list_remotes`."""
156
157 name: str # always present
158 url: str # always present
159
160 # ---------------------------------------------------------------------------
161 # Internal helpers
162 # ---------------------------------------------------------------------------
163
164 def _config_path(repo_root: pathlib.Path | None) -> pathlib.Path:
165 root = (repo_root or pathlib.Path.cwd()).resolve()
166 return _config_toml_path(root)
167
168 def _load_config(config_path: pathlib.Path) -> MuseConfig:
169 """Load and parse config.toml; return an empty MuseConfig if absent."""
170 if not config_path.is_file():
171 return {}
172
173 try:
174 with config_path.open("rb") as fh:
175 raw = tomllib.load(fh)
176 except Exception as exc: # noqa: BLE001
177 logger.warning("⚠️ Failed to parse %s: %s", config_path, exc)
178 return {}
179
180 config: MuseConfig = {}
181
182 hub_raw = raw.get("hub")
183 if isinstance(hub_raw, dict):
184 hub: HubConfig = {}
185 url_val = hub_raw.get("url")
186 if isinstance(url_val, str):
187 hub["url"] = url_val
188 config["hub"] = hub
189
190 remotes_raw = raw.get("remotes")
191 if isinstance(remotes_raw, dict):
192 remotes: RemotesMap = {}
193 for name, remote_raw in remotes_raw.items():
194 if isinstance(remote_raw, dict):
195 entry: RemoteEntry = {}
196 rurl = remote_raw.get("url")
197 if isinstance(rurl, str):
198 entry["url"] = rurl
199 branch_val = remote_raw.get("branch")
200 if isinstance(branch_val, str):
201 entry["branch"] = branch_val
202 promisor_val = remote_raw.get("promisor")
203 if isinstance(promisor_val, bool):
204 entry["promisor"] = promisor_val
205 remotes[name] = entry
206 config["remotes"] = remotes
207
208 domain_raw = raw.get("domain")
209 if isinstance(domain_raw, dict):
210 domain: DomainConfig = {}
211 for key, val in domain_raw.items():
212 if isinstance(val, str):
213 domain[key] = val
214 config["domain"] = domain
215
216 limits_raw = raw.get("limits")
217 if isinstance(limits_raw, dict):
218 limits: LimitsConfig = {}
219 mwc = limits_raw.get("max_walk_commits")
220 if isinstance(mwc, int) and mwc > 0:
221 limits["max_walk_commits"] = mwc
222 ma = limits_raw.get("max_ancestors")
223 if isinstance(ma, int) and ma > 0:
224 limits["max_ancestors"] = ma
225 mgc = limits_raw.get("max_graph_commits")
226 if isinstance(mgc, int) and mgc > 0:
227 limits["max_graph_commits"] = mgc
228 spl = limits_raw.get("shard_prefix_length")
229 if isinstance(spl, int) and spl in (2, 4):
230 limits["shard_prefix_length"] = spl
231 config["limits"] = limits
232
233 commit_raw = raw.get("commit")
234 if isinstance(commit_raw, dict):
235 commit_cfg: CommitConfig = {}
236 sign_v = commit_raw.get("sign")
237 if isinstance(sign_v, bool):
238 commit_cfg["sign"] = sign_v
239 if commit_cfg:
240 config["commit"] = commit_cfg
241
242 reflog_raw = raw.get("reflog")
243 if isinstance(reflog_raw, dict):
244 reflog_cfg: dict = {}
245 ed = reflog_raw.get("expire_days")
246 if isinstance(ed, int) and ed > 0:
247 reflog_cfg["expire_days"] = ed
248 if reflog_cfg:
249 config["reflog"] = reflog_cfg
250
251 symlog_raw = raw.get("symlog")
252 if isinstance(symlog_raw, dict):
253 symlog_cfg: SymlogConfig = {}
254 sl_ed = symlog_raw.get("expire_days")
255 if isinstance(sl_ed, int) and sl_ed > 0:
256 symlog_cfg["expire_days"] = sl_ed
257 if symlog_cfg:
258 config["symlog"] = symlog_cfg
259
260 push_raw = raw.get("push")
261 if isinstance(push_raw, dict):
262 push_cfg: PushConfig = {}
263 tags_v = push_raw.get("tags")
264 if isinstance(tags_v, bool):
265 push_cfg["tags"] = tags_v
266 if push_cfg:
267 config["push"] = push_cfg
268
269 branch_raw = raw.get("branch")
270 if isinstance(branch_raw, dict):
271 branch_map: dict[str, BranchMeta] = {}
272 for bname, bdata in branch_raw.items():
273 if not isinstance(bdata, dict):
274 continue
275 bmeta: BranchMeta = {}
276 intent_v = bdata.get("intent")
277 if isinstance(intent_v, str):
278 bmeta["intent"] = intent_v
279 resumable_v = bdata.get("resumable")
280 if isinstance(resumable_v, bool):
281 bmeta["resumable"] = resumable_v
282 remote_v = bdata.get("remote")
283 if isinstance(remote_v, str):
284 bmeta["remote"] = remote_v
285 merge_v = bdata.get("merge")
286 if isinstance(merge_v, str):
287 bmeta["merge"] = merge_v
288 branch_map[bname] = bmeta
289 if branch_map:
290 config["branch"] = branch_map
291
292 pb_raw = raw.get("protected_branches")
293 if isinstance(pb_raw, dict):
294 branches_val = pb_raw.get("branches")
295 if isinstance(branches_val, list):
296 patterns = [p for p in branches_val if isinstance(p, str)]
297 config["protected_branches"] = patterns
298
299 return config
300
301 def _escape(value: str) -> str:
302 """Escape a TOML basic string value (backslash and double-quote only).
303
304 TOML basic strings allow control characters escaped as ``\\n``, ``\\t``,
305 etc., but we store only printable content — control characters in values
306 are also stripped here so the resulting TOML file remains parseable.
307 """
308 return (
309 value.replace("\\", "\\\\")
310 .replace('"', '\\"')
311 .replace("\n", "\\n")
312 .replace("\r", "\\r")
313 .replace("\0", "")
314 )
315
316 # Characters that are structurally significant in unquoted TOML keys and
317 # table headers. Any of these in a key name would allow injection of
318 # arbitrary TOML sections or key-value pairs.
319 _TOML_KEY_UNSAFE: frozenset[str] = frozenset('\n\r\0][="')
320
321 def _validate_toml_key(key: str, context: str = "key") -> None:
322 """Raise ``ValueError`` if *key* contains TOML-structurally unsafe characters.
323
324 Prevents injection attacks where a crafted key like ``x]\\n[injected`` would
325 break the TOML section structure and allow writing arbitrary sections.
326
327 Args:
328 key: Key string to validate.
329 context: Human-readable label used in the error message (e.g. ``"domain key"``).
330
331 Raises:
332 ValueError: If any character in *key* is in ``_TOML_KEY_UNSAFE``.
333 """
334 bad = _TOML_KEY_UNSAFE & set(key)
335 if bad:
336 chars = ", ".join(sorted(repr(c) for c in bad))
337 raise ValueError(
338 f"Config {context} {key!r} contains characters not allowed in TOML keys: {chars}"
339 )
340
341 def _dump_toml(config: MuseConfig) -> str:
342 """Serialise a MuseConfig to TOML text.
343
344 Section order: ``[hub]``, ``[remotes.*]``, ``[domain]``, ``[limits]``.
345
346 All key names are validated against ``_TOML_KEY_UNSAFE`` before being
347 written, preventing TOML injection via crafted domain keys or remote names.
348 """
349 lines: list[str] = []
350
351 hub = config.get("hub")
352 if hub:
353 lines.append("[hub]")
354 url = hub.get("url", "")
355 if url:
356 lines.append(f'url = "{_escape(url)}"')
357 lines.append("")
358
359 remotes = config.get("remotes") or {}
360 for remote_name in sorted(remotes):
361 # Remote names come from _load_config which parses TOML, so they are
362 # safe at read time. Validate defensively before writing.
363 _validate_toml_key(remote_name, "remote name")
364 entry = remotes[remote_name]
365 lines.append(f"[remotes.{remote_name}]")
366 rurl = entry.get("url", "")
367 if rurl:
368 lines.append(f'url = "{_escape(rurl)}"')
369 branch = entry.get("branch", "")
370 if branch:
371 lines.append(f'branch = "{_escape(branch)}"')
372 if "promisor" in entry:
373 lines.append(f'promisor = {"true" if entry["promisor"] else "false"}')
374 lines.append("")
375
376 domain = config.get("domain") or {}
377 if domain:
378 lines.append("[domain]")
379 for key, val in sorted(domain.items()):
380 _validate_toml_key(key, "domain key")
381 lines.append(f'{key} = "{_escape(val)}"')
382 lines.append("")
383
384 limits = config.get("limits") or {}
385 if limits:
386 lines.append("[limits]")
387 mwc = limits.get("max_walk_commits")
388 if mwc is not None:
389 lines.append(f"max_walk_commits = {mwc}")
390 ma = limits.get("max_ancestors")
391 if ma is not None:
392 lines.append(f"max_ancestors = {ma}")
393 mgc = limits.get("max_graph_commits")
394 if mgc is not None:
395 lines.append(f"max_graph_commits = {mgc}")
396 spl = limits.get("shard_prefix_length")
397 if spl is not None:
398 lines.append(f"shard_prefix_length = {spl}")
399 lines.append("")
400
401 commit_cfg = config.get("commit") or {}
402 if commit_cfg:
403 lines.append("[commit]")
404 if "sign" in commit_cfg:
405 lines.append(f"sign = {'true' if commit_cfg['sign'] else 'false'}")
406 lines.append("")
407
408 reflog_cfg = config.get("reflog") or {}
409 if reflog_cfg:
410 lines.append("[reflog]")
411 ed = reflog_cfg.get("expire_days")
412 if ed is not None:
413 lines.append(f"expire_days = {ed}")
414 lines.append("")
415
416 symlog_cfg_out = config.get("symlog") or {}
417 if symlog_cfg_out:
418 lines.append("[symlog]")
419 sl_ed = symlog_cfg_out.get("expire_days")
420 if sl_ed is not None:
421 lines.append(f"expire_days = {sl_ed}")
422 lines.append("")
423
424 push_cfg_out = config.get("push") or {}
425 if push_cfg_out:
426 lines.append("[push]")
427 if "tags" in push_cfg_out:
428 lines.append(f"tags = {'true' if push_cfg_out['tags'] else 'false'}")
429 lines.append("")
430
431 branch_sections = config.get("branch") or {}
432 for bname in sorted(branch_sections):
433 _validate_toml_key(bname, "branch name")
434 bmeta = branch_sections[bname]
435 # Skip empty metadata dicts — no section needed.
436 if not bmeta:
437 continue
438 # Branch names require quoted keys (may contain '/' and other chars
439 # that are not valid in bare TOML keys).
440 lines.append(f'[branch."{_escape(bname)}"]')
441 intent = bmeta.get("intent", "")
442 if intent:
443 lines.append(f'intent = "{_escape(intent)}"')
444 if "resumable" in bmeta:
445 lines.append(f"resumable = {'true' if bmeta['resumable'] else 'false'}")
446 remote = bmeta.get("remote", "")
447 if remote:
448 lines.append(f'remote = "{_escape(remote)}"')
449 merge = bmeta.get("merge", "")
450 if merge:
451 lines.append(f'merge = "{_escape(merge)}"')
452 lines.append("")
453
454 return "\n".join(lines)
455
456 # ---------------------------------------------------------------------------
457 # Auth token resolution (via identity store)
458 # ---------------------------------------------------------------------------
459
460 def get_signing_identity(
461 repo_root: pathlib.Path | None = None,
462 remote_url: str | None = None,
463 agent_id: str | None = None,
464 ) -> "object | None":
465 """Return a :class:`~muse.core.transport.SigningIdentity` for a hub, or ``None``.
466
467 Resolution order:
468 1. ``MUSE_AGENT_KEY_FD`` environment variable — integer file descriptor
469 from which exactly 64 bytes of sub-seed are read (then the fd is
470 closed). The Ed25519 identity key is derived via
471 :func:`~muse.core.hdkeys.derive_identity_key`, scoped to whichever
472 hub is resolved from *remote_url* / ``[hub] url`` (musehub#221) — if
473 no hub is resolvable, this branch is skipped entirely rather than
474 deriving a hub-independent key. The handle is taken
475 from ``MUSE_AGENT_HANDLE`` (defaults to *agent_id* if set, else
476 ``"agent"``). This is the only supported env-based injection mechanism;
477 the secret travels through the kernel pipe buffer and never appears in
478 ``/proc/<pid>/environ``.
479 2. Agent-specific entry in ``~/.muse/identity.toml`` keyed by
480 ``"hostname#agent_id"`` — when *agent_id* is provided.
481 3. Human entry in ``~/.muse/identity.toml`` keyed by bare hostname.
482 4. Hub URL from ``[hub] url`` in ``.muse/config.toml`` (fallback lookup
483 URL when *remote_url* is not supplied).
484
485 The private key is **never** logged.
486
487 Args:
488 repo_root: Repository root. Defaults to ``Path.cwd()``.
489 remote_url: URL of the specific remote being contacted.
490 agent_id: Agent handle, e.g. ``"agentception-abc123"``. Used to
491 try an agent-specific key before falling back to the
492 human key.
493
494 Returns:
495 :class:`~muse.core.transport.SigningIdentity` or ``None``.
496 """
497 import os as _os
498 from muse.core.identity import resolve_signing_identity, hostname_from_url # avoid circular import
499 from muse.core.transport import SigningIdentity
500 from muse.core.hdkeys import hub_index
501
502 # Resolved once, up front — every derivation below is scoped to this hub
503 # (musehub#221). Without a resolvable hub, MUSE_AGENT_KEY_FD cannot
504 # safely derive a key: there is no legitimate hub-independent identity
505 # key, so we fail closed and fall through to the identity-store lookup
506 # (which will itself return None below if there's truly no hub context).
507 lookup_url: str | None = remote_url or get_hub_url(repo_root)
508
509 # 1. MUSE_AGENT_KEY_FD — read 64-byte sub-seed from a pipe fd.
510 # This is the only supported env-var injection mechanism.
511 # The secret travels through the kernel pipe buffer and never appears
512 # in /proc/<pid>/environ.
513 key_fd_str = _os.environ.get("MUSE_AGENT_KEY_FD", "").strip()
514 if key_fd_str and lookup_url is not None:
515 try:
516 key_fd = int(key_fd_str)
517 import os as _os2
518 sub_seed = bytearray(_os2.read(key_fd, 64))
519 _os2.close(key_fd)
520 if len(sub_seed) == 64:
521 from muse.core.hdkeys import derive_identity_key
522 hub = hub_index(hostname_from_url(lookup_url))
523 dk = derive_identity_key(sub_seed, hub=hub)
524 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
525 private_key = Ed25519PrivateKey.from_private_bytes(dk.private_bytes)
526 dk.zero()
527 sub_seed[:] = b"\x00" * len(sub_seed)
528 handle = (
529 _os.environ.get("MUSE_AGENT_HANDLE", "").strip()
530 or agent_id
531 or "agent"
532 )
533 logger.debug("✅ Signing identity from MUSE_AGENT_KEY_FD (handle=%s)", handle)
534 return SigningIdentity(handle=handle, private_key=private_key)
535 logger.warning(
536 "⚠️ MUSE_AGENT_KEY_FD fd=%s yielded %d bytes (expected 64) — falling through",
537 key_fd_str, len(sub_seed),
538 )
539 except Exception as exc:
540 logger.warning("⚠️ MUSE_AGENT_KEY_FD could not be read: %s — falling through", exc)
541 elif key_fd_str:
542 logger.warning(
543 "⚠️ MUSE_AGENT_KEY_FD set but no hub is resolvable (no remote_url, no "
544 "[hub] url) — cannot derive a hub-scoped key, falling through"
545 )
546
547 # 2. Identity store lookup (agent key → human key fallback).
548 if lookup_url is None:
549 logger.debug("⚠️ No hub configured — skipping signing identity lookup")
550 return None
551
552 result = resolve_signing_identity(lookup_url, agent_id=agent_id)
553 if result is None:
554 logger.debug(
555 "⚠️ No signing identity for hub %s — run `muse auth keygen && muse auth register`",
556 lookup_url,
557 )
558 return None
559
560 handle, private_key = result
561 logger.debug("✅ Signing identity resolved for hub %s (handle=%s)", lookup_url, handle)
562 return SigningIdentity(handle=handle, private_key=private_key)
563
564 # ---------------------------------------------------------------------------
565 # Hub helpers
566 # ---------------------------------------------------------------------------
567
568 def get_hub_url(repo_root: pathlib.Path | None = None) -> str | None:
569 """Return the hub URL from ``[hub] url``, or ``None`` if not configured.
570
571 Resolution order:
572 1. ``<repo>/.muse/config.toml`` — repo-local config (highest priority).
573 2. ``~/.muse/config.toml`` — global user config (fallback).
574
575 This fallback ensures ``muse auth whoami`` and other hub-aware commands
576 work without ``--hub`` even when invoked outside a repository, as long as
577 the user has set a default hub in their global config.
578
579 Args:
580 repo_root: Repository root. Defaults to ``Path.cwd()``.
581
582 Returns:
583 URL string, or ``None``.
584 """
585 config = _load_config(_config_path(repo_root))
586 hub = config.get("hub")
587 if hub is not None:
588 url = hub.get("url", "")
589 if url.strip():
590 return url.strip()
591
592 # Fall back to ~/.muse/config.toml so hub-aware commands (e.g. `muse auth
593 # whoami`) work without --hub when invoked outside a repository.
594 global_config = _load_config(_GLOBAL_CONFIG_FILE)
595 global_hub = global_config.get("hub")
596 if global_hub is not None:
597 url = global_hub.get("url", "")
598 if url.strip():
599 return url.strip()
600
601 return None
602
603 def set_hub_url(url: str, repo_root: pathlib.Path | None = None) -> None:
604 """Write ``[hub] url`` to ``.muse/config.toml``.
605
606 Preserves all other sections. Creates the config file if absent.
607 Rejects ``http://`` URLs — Muse never contacts a hub over cleartext HTTP.
608
609 Args:
610 url: Hub URL (must be ``https://``).
611 repo_root: Repository root. Defaults to ``Path.cwd()``.
612
613 Raises:
614 ValueError: If *url* does not use the ``https://`` scheme.
615 """
616 _is_loopback = url.startswith("http://localhost") or url.startswith("http://127.0.0.1") or url.startswith("http://[::1]")
617 if not url.startswith("https://") and not _is_loopback:
618 raise ValueError(
619 f"Hub URL must use HTTPS. Got: {url!r}\n"
620 "Muse never connects to a hub over cleartext HTTP.\n"
621 "(Exception: http://localhost and http://127.0.0.1 are allowed for local development.)"
622 )
623 cp = _config_path(repo_root)
624 cp.parent.mkdir(parents=True, exist_ok=True)
625 config = _load_config(cp)
626 config["hub"] = HubConfig(url=url)
627 write_text_atomic(cp, _dump_toml(config))
628 logger.info("✅ Hub URL set to %s", url)
629
630 def clear_hub_url(repo_root: pathlib.Path | None = None) -> None:
631 """Remove the ``[hub]`` section from ``.muse/config.toml``.
632
633 Args:
634 repo_root: Repository root. Defaults to ``Path.cwd()``.
635 """
636 cp = _config_path(repo_root)
637 config = _load_config(cp)
638 if "hub" in config:
639 del config["hub"]
640 write_text_atomic(cp, _dump_toml(config))
641 logger.info("✅ Hub disconnected")
642
643 # ---------------------------------------------------------------------------
644 # Generic dotted-key helpers
645 # ---------------------------------------------------------------------------
646
647 _BlockedNS = dict[str, str]
648 _BLOCKED_NAMESPACES: _BlockedNS = {
649 "auth": "Use `muse auth keygen` and `muse auth register` to manage credentials.",
650 "remotes": "Use `muse remote add/remove/rename` to manage remotes.",
651 "user": "User identity is managed via `muse auth register`. Run `muse auth whoami` to inspect.",
652 }
653
654 _SETTABLE_NAMESPACES = {"hub", "domain", "limits", "commit", "reflog", "symlog", "push"}
655
656 # Default cap values — used when the [limits] section is absent or the key
657 # is not set. These are the same values that were previously hardcoded inside
658 # the individual functions.
659 _DEFAULT_MAX_WALK_COMMITS: int = 10_000
660 _DEFAULT_MAX_ANCESTORS: int = 50_000
661 _DEFAULT_MAX_GRAPH_COMMITS: int = 50_000
662 _DEFAULT_SHARD_PREFIX_LENGTH: int = 2
663
664 def get_limit(key: str, repo_root: pathlib.Path | None = None) -> int:
665 """Return a ``[limits]`` integer cap from config, or its default.
666
667 Args:
668 key: Limit key — one of ``max_walk_commits``, ``max_ancestors``,
669 ``max_graph_commits``.
670 repo_root: Repository root; ``None`` falls back to ``Path.cwd()``.
671
672 Returns:
673 Configured integer value, or the built-in default if not set.
674 """
675 defaults: DefaultsMap = {
676 "max_walk_commits": _DEFAULT_MAX_WALK_COMMITS,
677 "max_ancestors": _DEFAULT_MAX_ANCESTORS,
678 "max_graph_commits": _DEFAULT_MAX_GRAPH_COMMITS,
679 "shard_prefix_length": _DEFAULT_SHARD_PREFIX_LENGTH,
680 }
681 default = defaults.get(key, 10_000)
682 config = _load_config(_config_path(repo_root))
683 limits = config.get("limits") or {}
684 # Explicit key dispatch keeps mypy happy on TypedDict literal-required keys.
685 if key == "max_walk_commits":
686 val: int | None = limits.get("max_walk_commits")
687 elif key == "max_ancestors":
688 val = limits.get("max_ancestors")
689 elif key == "max_graph_commits":
690 val = limits.get("max_graph_commits")
691 elif key == "shard_prefix_length":
692 val = limits.get("shard_prefix_length")
693 else:
694 val = None
695 if isinstance(val, int) and val > 0:
696 return val
697 return default
698
699 def get_config_value(key: str, repo_root: pathlib.Path | None = None) -> str | None:
700 """Get a config value by dotted key (e.g. ``user.handle``, ``hub.url``).
701
702 Returns ``None`` when the key is not set or the namespace is unknown.
703
704 Args:
705 key: Dotted key in ``<namespace>.<subkey>`` form.
706 repo_root: Repository root. Defaults to ``Path.cwd()``.
707
708 Returns:
709 String value, or ``None``.
710 """
711 parts = key.split(".", 1)
712 if len(parts) != 2:
713 return None
714 namespace, subkey = parts
715 config = _load_config(_config_path(repo_root))
716
717 if namespace == "user":
718 # User identity lives in identity.toml, keyed by the configured hub URL.
719 hub_url = (config.get("hub") or {}).get("url", "")
720 if not hub_url:
721 return None
722 try:
723 from muse.core.identity import load_identity, hostname_from_url
724 hostname = hostname_from_url(hub_url)
725 entry = load_identity(hostname)
726 if entry is None:
727 return None
728 if subkey == "handle":
729 return entry.get("handle")
730 if subkey == "type":
731 return entry.get("type")
732 if subkey == "display_name":
733 return entry.get("display_name")
734 if subkey == "email":
735 return entry.get("email")
736 except Exception:
737 pass
738 return None
739
740 if namespace == "hub":
741 hub = config.get("hub") or {}
742 if subkey == "url":
743 return hub.get("url")
744 return None
745
746 if namespace == "domain":
747 domain = config.get("domain") or {}
748 return domain.get(subkey)
749
750 if namespace == "limits":
751 limits = config.get("limits") or {}
752 if subkey == "max_walk_commits":
753 v = limits.get("max_walk_commits")
754 return str(v) if isinstance(v, int) else None
755 if subkey == "max_ancestors":
756 v = limits.get("max_ancestors")
757 return str(v) if isinstance(v, int) else None
758 if subkey == "max_graph_commits":
759 v = limits.get("max_graph_commits")
760 return str(v) if isinstance(v, int) else None
761 if subkey == "shard_prefix_length":
762 v = limits.get("shard_prefix_length")
763 return str(v) if isinstance(v, int) else None
764 return None
765
766 if namespace == "commit":
767 commit = config.get("commit") or {}
768 if subkey == "sign":
769 v = commit.get("sign")
770 if v is True:
771 return "true"
772 if v is False:
773 return "false"
774 return None
775 return None
776
777 if namespace == "reflog":
778 reflog = config.get("reflog") or {}
779 if subkey == "expire-days":
780 v = reflog.get("expire_days")
781 return str(v) if isinstance(v, int) else None
782 return None
783
784 if namespace == "symlog":
785 symlog = config.get("symlog") or {}
786 if subkey == "expire-days":
787 v = symlog.get("expire_days")
788 return str(v) if isinstance(v, int) else None
789 return None
790
791 if namespace == "push":
792 push = config.get("push") or {}
793 if subkey == "tags":
794 v = push.get("tags")
795 if v is True:
796 return "true"
797 if v is False:
798 return "false"
799 return None
800 return None
801
802 return None
803
804 def set_config_value(key: str, value: str, repo_root: pathlib.Path | None = None) -> None:
805 """Set a config value by dotted key (e.g. ``user.handle``, ``domain.ticks_per_beat``).
806
807 Args:
808 key: Dotted key in ``<namespace>.<subkey>`` form.
809 value: New string value.
810 repo_root: Repository root. Defaults to ``Path.cwd()``.
811
812 Raises:
813 ValueError: If the namespace is blocked, unknown, or the subkey is invalid.
814 """
815 parts = key.split(".", 1)
816 if len(parts) != 2:
817 raise ValueError(f"Key must be in 'namespace.subkey' form, got: {key!r}")
818 namespace, subkey = parts
819
820 if namespace in _BLOCKED_NAMESPACES:
821 raise ValueError(_BLOCKED_NAMESPACES[namespace])
822
823 if namespace not in _SETTABLE_NAMESPACES:
824 raise ValueError(
825 f"Unknown config namespace {namespace!r}. "
826 f"Settable namespaces: {', '.join(sorted(_SETTABLE_NAMESPACES))}"
827 )
828
829 cp = _config_path(repo_root)
830 cp.parent.mkdir(parents=True, exist_ok=True)
831 config = _load_config(cp)
832
833 if namespace == "user":
834 set_user_field(subkey, value, repo_root)
835 return
836
837 if namespace == "hub":
838 if subkey != "url":
839 raise ValueError(f"Unknown [hub] config key: {subkey!r}. Valid keys: url")
840 # Route through set_hub_url — it enforces the HTTPS requirement.
841 set_hub_url(value, repo_root)
842 return
843
844 if namespace == "limits":
845 _LIMITS_KEYS = frozenset({
846 "max_walk_commits", "max_ancestors", "max_graph_commits", "shard_prefix_length",
847 })
848 if subkey not in _LIMITS_KEYS:
849 raise ValueError(
850 f"Unknown [limits] config key: {subkey!r}. "
851 f"Valid keys: {', '.join(sorted(_LIMITS_KEYS))}"
852 )
853 try:
854 int_value = int(value)
855 except ValueError as exc:
856 raise ValueError(
857 f"[limits] {subkey} must be an integer, got: {value!r}"
858 ) from exc
859 if int_value <= 0:
860 raise ValueError(f"[limits] {subkey} must be a positive integer, got: {int_value}")
861 if subkey == "shard_prefix_length" and int_value not in (2, 4):
862 raise ValueError("shard_prefix_length must be 2 or 4")
863 limits_section: LimitsConfig = config.get("limits") or {}
864 if subkey == "max_walk_commits":
865 limits_section["max_walk_commits"] = int_value
866 elif subkey == "max_ancestors":
867 limits_section["max_ancestors"] = int_value
868 elif subkey == "max_graph_commits":
869 limits_section["max_graph_commits"] = int_value
870 elif subkey == "shard_prefix_length":
871 limits_section["shard_prefix_length"] = int_value
872 config["limits"] = limits_section
873 write_text_atomic(cp, _dump_toml(config))
874 logger.info("✅ limits.%s = %d", subkey, int_value)
875 return
876
877 if namespace == "commit":
878 _COMMIT_KEYS = frozenset({"sign"})
879 if subkey not in _COMMIT_KEYS:
880 raise ValueError(
881 f"Unknown [commit] config key: {subkey!r}. "
882 f"Valid keys: {', '.join(sorted(_COMMIT_KEYS))}"
883 )
884 if value not in ("true", "false"):
885 raise ValueError(f"[commit] {subkey} must be 'true' or 'false', got: {value!r}")
886 commit_section: CommitConfig = config.get("commit") or {}
887 if subkey == "sign":
888 commit_section["sign"] = value == "true"
889 config["commit"] = commit_section
890 write_text_atomic(cp, _dump_toml(config))
891 logger.info("✅ commit.%s = %s", subkey, value)
892 return
893
894 if namespace == "reflog":
895 _REFLOG_KEYS = frozenset({"expire-days"})
896 if subkey not in _REFLOG_KEYS:
897 raise ValueError(
898 f"Unknown [reflog] config key: {subkey!r}. "
899 f"Valid keys: {', '.join(sorted(_REFLOG_KEYS))}"
900 )
901 if subkey == "expire-days":
902 try:
903 int_value = int(value)
904 except ValueError as exc:
905 raise ValueError(
906 f"[reflog] expire-days must be a positive integer, got: {value!r}"
907 ) from exc
908 if int_value <= 0:
909 raise ValueError(f"[reflog] expire-days must be a positive integer, got: {int_value}")
910 reflog_section: dict = config.get("reflog") or {}
911 reflog_section["expire_days"] = int_value
912 config["reflog"] = reflog_section
913 write_text_atomic(cp, _dump_toml(config))
914 logger.info("✅ reflog.expire-days = %d", int_value)
915 return
916
917 if namespace == "symlog":
918 _SYMLOG_KEYS = frozenset({"expire-days"})
919 if subkey not in _SYMLOG_KEYS:
920 raise ValueError(
921 f"Unknown [symlog] config key: {subkey!r}. "
922 f"Valid keys: {', '.join(sorted(_SYMLOG_KEYS))}"
923 )
924 if subkey == "expire-days":
925 try:
926 sl_int_value = int(value)
927 except ValueError as exc:
928 raise ValueError(
929 f"[symlog] expire-days must be a positive integer, got: {value!r}"
930 ) from exc
931 if sl_int_value <= 0:
932 raise ValueError(f"[symlog] expire-days must be a positive integer, got: {sl_int_value}")
933 symlog_section: SymlogConfig = config.get("symlog") or {}
934 symlog_section["expire_days"] = sl_int_value
935 config["symlog"] = symlog_section
936 write_text_atomic(cp, _dump_toml(config))
937 logger.info("✅ symlog.expire-days = %d", sl_int_value)
938 return
939
940 if namespace == "push":
941 _PUSH_KEYS = frozenset({"tags"})
942 if subkey not in _PUSH_KEYS:
943 raise ValueError(
944 f"Unknown [push] config key: {subkey!r}. "
945 f"Valid keys: {', '.join(sorted(_PUSH_KEYS))}"
946 )
947 if value not in ("true", "false"):
948 raise ValueError(f"[push] {subkey} must be 'true' or 'false', got: {value!r}")
949 push_section: dict = config.get("push") or {}
950 push_section[subkey] = value == "true"
951 config["push"] = push_section
952 write_text_atomic(cp, _dump_toml(config))
953 logger.info("✅ push.%s = %s", subkey, value)
954 return
955
956 # namespace == "domain"
957 _validate_toml_key(subkey, "domain key")
958 domain: DomainConfig = config.get("domain") or {}
959 domain[subkey] = value
960 config["domain"] = domain
961 write_text_atomic(cp, _dump_toml(config))
962 logger.info("✅ domain.%s = %r", subkey, value)
963
964 def config_as_dict(repo_root: pathlib.Path | None = None) -> ConfigTree:
965 """Return the full config as a plain ``dict[str, dict[str, str]]`` for JSON output.
966
967 Credentials are never included — the hub section only contains the URL.
968
969 Args:
970 repo_root: Repository root. Defaults to ``Path.cwd()``.
971
972 Returns:
973 Nested dict suitable for ``json.dumps``.
974 """
975 config = _load_config(_config_path(repo_root))
976 result: ConfigTree = {}
977
978 hub = config.get("hub")
979 if hub:
980 hub_url = hub.get("url", "")
981 if hub_url:
982 result["hub"] = {"url": hub_url}
983
984 remotes = config.get("remotes") or {}
985 if remotes:
986 remotes_dict: ConfigSection = {}
987 for rname, entry in sorted(remotes.items()):
988 url = entry.get("url", "")
989 if url:
990 remotes_dict[rname] = url
991 if remotes_dict:
992 result["remotes"] = remotes_dict
993
994 domain = config.get("domain") or {}
995 if domain:
996 result["domain"] = dict(sorted(domain.items()))
997
998 limits = config.get("limits") or {}
999 if limits:
1000 limits_dict: ConfigSection = {}
1001 for lk in ("max_walk_commits", "max_ancestors", "max_graph_commits", "shard_prefix_length"):
1002 lv = limits.get(lk)
1003 if lv is not None:
1004 limits_dict[lk] = str(lv)
1005 if limits_dict:
1006 result["limits"] = limits_dict
1007
1008 return result
1009
1010 def config_path_for_editor(repo_root: pathlib.Path | None = None) -> pathlib.Path:
1011 """Return the config path for the ``config edit`` command."""
1012 return _config_path(repo_root)
1013
1014 # ---------------------------------------------------------------------------
1015 # Branch metadata helpers
1016 # ---------------------------------------------------------------------------
1017
1018 def write_branch_meta(
1019 repo_root: pathlib.Path,
1020 branch_name: str,
1021 *,
1022 intent: str | None = None,
1023 resumable: bool | None = None,
1024 ) -> None:
1025 """Write per-branch metadata to ``[branch."<name>"]`` in ``.muse/config.toml``.
1026
1027 Only the supplied keyword arguments are updated; existing fields
1028 (``remote``, ``merge``, and previously written ``intent``/``resumable``)
1029 are preserved unchanged.
1030
1031 Args:
1032 repo_root: Repository root directory.
1033 branch_name: Name of the branch (e.g. ``"feat/my-thing"``).
1034 intent: Short description of what this branch is for.
1035 resumable: Mark this branch as a resumable agent checkpoint.
1036 """
1037 _validate_toml_key(branch_name, "branch name")
1038 cp = _config_path(repo_root)
1039 cp.parent.mkdir(parents=True, exist_ok=True)
1040 config = _load_config(cp)
1041 branch_map: dict[str, BranchMeta] = dict(config.get("branch") or {})
1042 entry: BranchMeta = dict(branch_map.get(branch_name) or {}) # type: ignore[arg-type]
1043 if intent is not None:
1044 entry["intent"] = intent
1045 if resumable is not None:
1046 entry["resumable"] = resumable
1047 branch_map[branch_name] = entry
1048 config["branch"] = branch_map
1049 write_text_atomic(cp, _dump_toml(config))
1050
1051 def delete_branch_meta(repo_root: pathlib.Path, branch_name: str) -> None:
1052 """Remove the ``[branch."<name>"]`` section from ``.muse/config.toml``.
1053
1054 Called by ``muse branch -d/-D`` after a successful branch deletion so
1055 stale intent/resumable entries do not accumulate indefinitely. No-op
1056 when the branch has no metadata or the config file is absent.
1057 """
1058 cp = _config_path(repo_root)
1059 if not cp.exists():
1060 return
1061 config = _load_config(cp)
1062 branch_map = dict(config.get("branch") or {})
1063 if branch_name not in branch_map:
1064 return
1065 del branch_map[branch_name]
1066 config["branch"] = branch_map
1067 write_text_atomic(cp, _dump_toml(config))
1068
1069 def read_branch_meta(
1070 repo_root: pathlib.Path,
1071 branch_name: str,
1072 ) -> BranchMeta:
1073 """Return per-branch metadata from ``.muse/config.toml``.
1074
1075 Returns an empty dict when the branch has no metadata or the config file
1076 is absent.
1077
1078 Args:
1079 repo_root: Repository root directory.
1080 branch_name: Name of the branch (e.g. ``"feat/my-thing"``).
1081
1082 Returns:
1083 Dict with any of: ``intent`` (str), ``resumable`` (bool),
1084 ``remote`` (str), ``merge`` (str).
1085 """
1086 config = _load_config(_config_path(repo_root))
1087 branch_map = config.get("branch") or {}
1088 return dict(branch_map.get(branch_name) or {})
1089
1090 # ---------------------------------------------------------------------------
1091 # Protected branches helpers
1092 # ---------------------------------------------------------------------------
1093
1094 def get_protected_branches(repo_root: pathlib.Path | None = None) -> list[str]:
1095 """Return the list of protected branch patterns from ``[protected_branches]``.
1096
1097 Returns an empty list when the section is absent or has no ``branches`` key.
1098
1099 Args:
1100 repo_root: Repository root. Defaults to ``Path.cwd()``.
1101 """
1102 config = _load_config(_config_path(repo_root))
1103 return list(config.get("protected_branches") or [])
1104
1105 def is_branch_protected(branch: str, patterns: list[str]) -> bool:
1106 """Return ``True`` if *branch* matches any pattern in *patterns*.
1107
1108 Patterns are matched with :func:`fnmatch.fnmatch` (shell-style globs).
1109 Matching is case-sensitive, consistent with Python fnmatch behaviour.
1110
1111 Args:
1112 branch: Branch name to test (e.g. ``"release/1.0"``).
1113 patterns: List of patterns from ``[protected_branches] branches``.
1114 """
1115 return any(fnmatch.fnmatch(branch, p) for p in patterns)
1116
1117 # ---------------------------------------------------------------------------
1118 # Remote helpers
1119 # ---------------------------------------------------------------------------
1120
1121 def get_remote(name: str, repo_root: pathlib.Path | None = None) -> str | None:
1122 """Return the URL for remote *name*, or ``None`` when not configured.
1123
1124 Args:
1125 name: Remote name (e.g. ``"origin"``).
1126 repo_root: Repository root. Defaults to ``Path.cwd()``.
1127
1128 Returns:
1129 URL string, or ``None``.
1130 """
1131 config = _load_config(_config_path(repo_root))
1132 remotes = config.get("remotes")
1133 if remotes is None:
1134 return None
1135 entry = remotes.get(name)
1136 if entry is None:
1137 return None
1138 url = entry.get("url", "")
1139 return url.strip() if url.strip() else None
1140
1141 def set_remote(
1142 name: str,
1143 url: str,
1144 repo_root: pathlib.Path | None = None,
1145 ) -> None:
1146 """Write ``[remotes.<name>] url`` to ``.muse/config.toml``.
1147
1148 Preserves all other sections. Creates the file if absent.
1149
1150 Args:
1151 name: Remote name (e.g. ``"origin"``).
1152 url: Remote URL.
1153 repo_root: Repository root. Defaults to ``Path.cwd()``.
1154 """
1155 cp = _config_path(repo_root)
1156 cp.parent.mkdir(parents=True, exist_ok=True)
1157 config = _load_config(cp)
1158 existing_remotes = config.get("remotes")
1159 remotes: RemotesMap = {}
1160 if existing_remotes:
1161 remotes.update(existing_remotes)
1162 existing_entry = remotes.get(name)
1163 entry: RemoteEntry = {}
1164 if existing_entry is not None:
1165 if "url" in existing_entry:
1166 entry["url"] = existing_entry["url"]
1167 if "branch" in existing_entry:
1168 entry["branch"] = existing_entry["branch"]
1169 entry["url"] = url
1170 remotes[name] = entry
1171 config["remotes"] = remotes
1172 write_text_atomic(cp, _dump_toml(config))
1173 logger.info("✅ Remote %r set to %s", name, url)
1174
1175 def remove_remote(
1176 name: str,
1177 repo_root: pathlib.Path | None = None,
1178 ) -> None:
1179 """Remove a named remote and its tracking refs.
1180
1181 Args:
1182 name: Remote name to remove.
1183 repo_root: Repository root. Defaults to ``Path.cwd()``.
1184
1185 Raises:
1186 KeyError: If *name* is not a configured remote.
1187 """
1188 cp = _config_path(repo_root)
1189 config = _load_config(cp)
1190 remotes = config.get("remotes")
1191 if remotes is None or name not in remotes:
1192 raise KeyError(name)
1193 del remotes[name]
1194 config["remotes"] = remotes
1195 write_text_atomic(cp, _dump_toml(config))
1196 logger.info("✅ Remote %r removed from config", name)
1197
1198 root = (repo_root or pathlib.Path.cwd()).resolve()
1199 refs_dir = _remote_tracking_dir(root, name)
1200 if refs_dir.is_symlink():
1201 # Refuse to rmtree a symlink — following a symlink placed by an
1202 # attacker could delete files outside the repository tree.
1203 logger.warning("⚠️ Skipping rmtree: remotes dir %s is a symlink", refs_dir)
1204 elif refs_dir.is_dir():
1205 shutil.rmtree(refs_dir)
1206 logger.debug("✅ Removed tracking refs dir %s", refs_dir)
1207
1208 def rename_remote(
1209 old_name: str,
1210 new_name: str,
1211 repo_root: pathlib.Path | None = None,
1212 ) -> None:
1213 """Rename a remote and move its tracking refs.
1214
1215 Args:
1216 old_name: Current remote name.
1217 new_name: Desired new remote name.
1218 repo_root: Repository root. Defaults to ``Path.cwd()``.
1219
1220 Raises:
1221 KeyError: If *old_name* is not a configured remote.
1222 ValueError: If *new_name* is already configured.
1223 """
1224 cp = _config_path(repo_root)
1225 config = _load_config(cp)
1226 remotes = config.get("remotes")
1227 if remotes is None or old_name not in remotes:
1228 raise KeyError(old_name)
1229 if new_name in remotes:
1230 raise ValueError(new_name)
1231 remotes[new_name] = remotes.pop(old_name)
1232 config["remotes"] = remotes
1233 write_text_atomic(cp, _dump_toml(config))
1234 logger.info("✅ Remote %r renamed to %r", old_name, new_name)
1235
1236 root = (repo_root or pathlib.Path.cwd()).resolve()
1237 old_refs_dir = _remote_tracking_dir(root, old_name)
1238 new_refs_dir = _remote_tracking_dir(root, new_name)
1239 if old_refs_dir.is_dir():
1240 old_refs_dir.rename(new_refs_dir)
1241 logger.debug("✅ Moved tracking refs dir %s → %s", old_refs_dir, new_refs_dir)
1242
1243 def list_remotes(repo_root: pathlib.Path | None = None) -> list[RemoteConfig]:
1244 """Return all configured remotes sorted alphabetically by name.
1245
1246 Args:
1247 repo_root: Repository root. Defaults to ``Path.cwd()``.
1248
1249 Returns:
1250 List of ``{"name": str, "url": str}`` dicts.
1251 """
1252 config = _load_config(_config_path(repo_root))
1253 remotes = config.get("remotes")
1254 if remotes is None:
1255 return []
1256 result: list[RemoteConfig] = []
1257 for remote_name in sorted(remotes):
1258 entry = remotes[remote_name]
1259 url = entry.get("url", "")
1260 if not url.strip():
1261 continue
1262 rc = RemoteConfig(name=remote_name, url=url.strip())
1263 result.append(rc)
1264 return result
1265
1266 # ---------------------------------------------------------------------------
1267 # Remote tracking-head helpers
1268 # ---------------------------------------------------------------------------
1269
1270 def _remote_head_path(
1271 remote_name: str,
1272 branch: str,
1273 repo_root: pathlib.Path | None = None,
1274 ) -> pathlib.Path:
1275 """Return the path to the remote tracking pointer file."""
1276 root = (repo_root or pathlib.Path.cwd()).resolve()
1277 return _remote_ref_path(root, remote_name, branch)
1278
1279 def get_remote_head(
1280 remote_name: str,
1281 branch: str,
1282 repo_root: pathlib.Path | None = None,
1283 ) -> str | None:
1284 """Return the last-known remote commit ID for *remote_name*/*branch*.
1285
1286 Returns ``None`` when the tracking pointer does not exist.
1287
1288 Args:
1289 remote_name: Remote name (e.g. ``"origin"``).
1290 branch: Branch name (e.g. ``"main"``).
1291 repo_root: Repository root. Defaults to ``Path.cwd()``.
1292
1293 Returns:
1294 Commit ID string, or ``None``.
1295 """
1296 return read_ref(_remote_head_path(remote_name, branch, repo_root))
1297
1298 def set_remote_head(
1299 remote_name: str,
1300 branch: str,
1301 commit_id: str,
1302 repo_root: pathlib.Path | None = None,
1303 ) -> None:
1304 """Write the remote tracking pointer for *remote_name*/*branch*.
1305
1306 Args:
1307 remote_name: Remote name (e.g. ``"origin"``).
1308 branch: Branch name.
1309 commit_id: Commit ID to record as the known remote HEAD.
1310 repo_root: Repository root. Defaults to ``Path.cwd()``.
1311 """
1312 pointer = _remote_head_path(remote_name, branch, repo_root)
1313 write_text_atomic(pointer, commit_id)
1314 logger.debug("✅ Remote head %s/%s → %s", remote_name, branch, short_id(commit_id))
1315
1316 def delete_remote_head(
1317 remote_name: str,
1318 branch: str,
1319 repo_root: pathlib.Path | None = None,
1320 ) -> bool:
1321 """Remove the local remote-tracking pointer for *remote_name*/*branch*.
1322
1323 Used after ``muse push --delete`` deletes the branch on the server, or when
1324 pruning stale tracking refs with ``muse branch -dr``.
1325
1326 Args:
1327 remote_name: Remote name (e.g. ``"origin"``).
1328 branch: Branch name (e.g. ``"feat/my-thing"``).
1329 repo_root: Repository root. Defaults to ``Path.cwd()``.
1330
1331 Returns:
1332 ``True`` if the pointer file existed and was removed, ``False`` if it
1333 was already absent (idempotent).
1334 """
1335 pointer = _remote_head_path(remote_name, branch, repo_root)
1336 if not pointer.is_file():
1337 return False
1338 pointer.unlink()
1339 # Remove now-empty parent directories (mirrors _cleanup_empty_dirs in branch.py).
1340 remotes_dir = pointer.parent
1341 while remotes_dir.name != remote_name:
1342 try:
1343 remotes_dir.rmdir()
1344 except OSError:
1345 break
1346 remotes_dir = remotes_dir.parent
1347 logger.debug("🗑 Remote tracking ref %s/%s removed", remote_name, branch)
1348 return True
1349
1350 # ---------------------------------------------------------------------------
1351 # Upstream tracking helpers
1352 # ---------------------------------------------------------------------------
1353
1354 def set_upstream(
1355 branch: str,
1356 remote_name: str,
1357 repo_root: pathlib.Path | None = None,
1358 ) -> None:
1359 """Record *remote_name* as the upstream remote for *branch*.
1360
1361 Args:
1362 branch: Local (and remote) branch name.
1363 remote_name: Remote name.
1364 repo_root: Repository root. Defaults to ``Path.cwd()``.
1365 """
1366 cp = _config_path(repo_root)
1367 cp.parent.mkdir(parents=True, exist_ok=True)
1368 config = _load_config(cp)
1369 existing_remotes = config.get("remotes")
1370 remotes: RemotesMap = {}
1371 if existing_remotes:
1372 remotes.update(existing_remotes)
1373 existing_entry = remotes.get(remote_name)
1374 entry: RemoteEntry = {}
1375 if existing_entry is not None:
1376 if "url" in existing_entry:
1377 entry["url"] = existing_entry["url"]
1378 if "branch" in existing_entry:
1379 entry["branch"] = existing_entry["branch"]
1380 entry["branch"] = branch
1381 remotes[remote_name] = entry
1382 config["remotes"] = remotes
1383 write_text_atomic(cp, _dump_toml(config))
1384 logger.info("✅ Upstream for branch %r set to %s/%r", branch, remote_name, branch)
1385
1386 def get_upstream(
1387 branch: str,
1388 repo_root: pathlib.Path | None = None,
1389 ) -> str | None:
1390 """Return the configured upstream remote name for *branch*, or ``None``.
1391
1392 Args:
1393 branch: Local branch name.
1394 repo_root: Repository root. Defaults to ``Path.cwd()``.
1395
1396 Returns:
1397 Remote name string, or ``None``.
1398 """
1399 config = _load_config(_config_path(repo_root))
1400 remotes = config.get("remotes")
1401 if remotes is None:
1402 return None
1403 for rname, entry in remotes.items():
1404 tracked = entry.get("branch", "")
1405 if tracked.strip() == branch:
1406 return rname
1407 return None
1408
1409 # ---------------------------------------------------------------------------
1410 # Global user config — ~/.muse/config.toml (safe_dirs)
1411 # ---------------------------------------------------------------------------
1412
1413 _GLOBAL_MUSE_DIR = _user_muse_dir()
1414 _GLOBAL_CONFIG_FILE = _user_config_toml_path()
1415
1416 def _load_global_config() -> _SecurityConfig:
1417 """Load ``~/.muse/config.toml`` and return the ``[security]`` section.
1418
1419 Returns a dict with key ``safe_dirs`` mapping to a list of path strings.
1420 Returns ``{"safe_dirs": []}`` when the file is absent or unparseable.
1421 """
1422 if not _GLOBAL_CONFIG_FILE.is_file():
1423 return {"safe_dirs": []}
1424 try:
1425 with _GLOBAL_CONFIG_FILE.open("rb") as fh:
1426 raw = tomllib.load(fh)
1427 except Exception as exc: # noqa: BLE001
1428 logger.warning("⚠️ Failed to parse %s: %s", _GLOBAL_CONFIG_FILE, exc)
1429 return {"safe_dirs": []}
1430 security_raw = raw.get("security")
1431 if not isinstance(security_raw, dict):
1432 return {"safe_dirs": []}
1433 dirs_raw = security_raw.get("safe_dirs")
1434 if not isinstance(dirs_raw, list):
1435 return {"safe_dirs": []}
1436 safe: list[str] = [d for d in dirs_raw if isinstance(d, str) and d.strip()]
1437 return {"safe_dirs": safe}
1438
1439 def _save_global_config(safe_dirs: list[str]) -> None:
1440 """Write ``[security] safe_dirs`` to ``~/.muse/config.toml``.
1441
1442 Preserves any other sections that may exist in the file.
1443 """
1444 import os
1445 _GLOBAL_MUSE_DIR.mkdir(parents=True, exist_ok=True)
1446
1447 # Read existing raw content to preserve other sections.
1448 existing_lines: list[str] = []
1449 if _GLOBAL_CONFIG_FILE.is_file():
1450 try:
1451 existing_lines = _GLOBAL_CONFIG_FILE.read_text("utf-8").splitlines()
1452 except Exception: # noqa: BLE001
1453 existing_lines = []
1454
1455 # Strip any existing [security] section from the file.
1456 filtered: list[str] = []
1457 in_security = False
1458 for line in existing_lines:
1459 stripped = line.strip()
1460 if stripped == "[security]":
1461 in_security = True
1462 continue
1463 if in_security and stripped.startswith("["):
1464 in_security = False
1465 if not in_security:
1466 filtered.append(line)
1467
1468 # Remove trailing blank lines before appending the new section.
1469 while filtered and not filtered[-1].strip():
1470 filtered.pop()
1471
1472 # Append the new [security] section.
1473 filtered.append("")
1474 filtered.append("[security]")
1475 if safe_dirs:
1476 items = ", ".join(f'"{_escape(d)}"' for d in safe_dirs)
1477 filtered.append(f"safe_dirs = [{items}]")
1478 else:
1479 filtered.append("safe_dirs = []")
1480 filtered.append("")
1481
1482 content = "\n".join(filtered)
1483 tmp = _GLOBAL_CONFIG_FILE.with_suffix(".toml.tmp")
1484 tmp.write_text(content, encoding="utf-8")
1485 os.replace(tmp, _GLOBAL_CONFIG_FILE)
1486
1487 def get_global_safe_dirs() -> list[str]:
1488 """Return the ``safe_dirs`` list from ``~/.muse/config.toml``.
1489
1490 Returns an empty list when not configured.
1491 """
1492 return _load_global_config().get("safe_dirs", [])
1493
1494 def add_global_safe_dir(path: str) -> None:
1495 """Add *path* to the ``safe_dirs`` list in ``~/.muse/config.toml``.
1496
1497 Normalises the path (``os.path.abspath``) before storing. Idempotent —
1498 adding the same path twice has no effect.
1499
1500 Args:
1501 path: Absolute or relative path to trust.
1502 """
1503 import os
1504 abs_path = os.path.abspath(path)
1505 current = get_global_safe_dirs()
1506 if abs_path not in current:
1507 current.append(abs_path)
1508 _save_global_config(current)
1509 logger.info("✅ Trusted path added: %s", abs_path)
1510
1511 def remove_global_safe_dir(path: str) -> None:
1512 """Remove *path* from the ``safe_dirs`` list in ``~/.muse/config.toml``.
1513
1514 Normalises the path before matching. No-op when the path is not present.
1515
1516 Args:
1517 path: Absolute or relative path to remove from the trust list.
1518 """
1519 import os
1520 abs_path = os.path.abspath(path)
1521 current = get_global_safe_dirs()
1522 updated = [d for d in current if d != abs_path]
1523 _save_global_config(updated)
1524 logger.info("✅ Trusted path removed: %s", abs_path)
File History 1 commit
sha256:94f494a8f59e4b708ebb89e304209737ce34324a33aae83840a6f6b06d7b8d9d docs: add domain-extensibility.md — the two-axis breadth/de… Sonnet 5 3 days ago