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