gabriel / muse public
config.py python
1,508 lines 54.0 KB
Raw
sha256:c0eba5ad689cec79f4a3fcdc4f5da78556cb4b8cb7b330f944634356c10379ed chore: pivot to nightly channel — bump version to 0.2.0.dev… Sonnet 5 patch 13 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`. The handle is taken
472 from ``MUSE_AGENT_HANDLE`` (defaults to *agent_id* if set, else
473 ``"agent"``). This is the only supported env-based injection mechanism;
474 the secret travels through the kernel pipe buffer and never appears in
475 ``/proc/<pid>/environ``.
476 2. Agent-specific entry in ``~/.muse/identity.toml`` keyed by
477 ``"hostname#agent_id"`` — when *agent_id* is provided.
478 3. Human entry in ``~/.muse/identity.toml`` keyed by bare hostname.
479 4. Hub URL from ``[hub] url`` in ``.muse/config.toml`` (fallback lookup
480 URL when *remote_url* is not supplied).
481
482 The private key is **never** logged.
483
484 Args:
485 repo_root: Repository root. Defaults to ``Path.cwd()``.
486 remote_url: URL of the specific remote being contacted.
487 agent_id: Agent handle, e.g. ``"agentception-abc123"``. Used to
488 try an agent-specific key before falling back to the
489 human key.
490
491 Returns:
492 :class:`~muse.core.transport.SigningIdentity` or ``None``.
493 """
494 import os as _os
495 from muse.core.identity import resolve_signing_identity # avoid circular import
496 from muse.core.transport import SigningIdentity
497
498 # 1. MUSE_AGENT_KEY_FD — read 64-byte sub-seed from a pipe fd.
499 # This is the only supported env-var injection mechanism.
500 # The secret travels through the kernel pipe buffer and never appears
501 # in /proc/<pid>/environ.
502 key_fd_str = _os.environ.get("MUSE_AGENT_KEY_FD", "").strip()
503 if key_fd_str:
504 try:
505 key_fd = int(key_fd_str)
506 import os as _os2
507 sub_seed = bytearray(_os2.read(key_fd, 64))
508 _os2.close(key_fd)
509 if len(sub_seed) == 64:
510 from muse.core.hdkeys import derive_identity_key
511 dk = derive_identity_key(sub_seed)
512 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
513 private_key = Ed25519PrivateKey.from_private_bytes(dk.private_bytes)
514 dk.zero()
515 sub_seed[:] = b"\x00" * len(sub_seed)
516 handle = (
517 _os.environ.get("MUSE_AGENT_HANDLE", "").strip()
518 or agent_id
519 or "agent"
520 )
521 logger.debug("✅ Signing identity from MUSE_AGENT_KEY_FD (handle=%s)", handle)
522 return SigningIdentity(handle=handle, private_key=private_key)
523 logger.warning(
524 "⚠️ MUSE_AGENT_KEY_FD fd=%s yielded %d bytes (expected 64) — falling through",
525 key_fd_str, len(sub_seed),
526 )
527 except Exception as exc:
528 logger.warning("⚠️ MUSE_AGENT_KEY_FD could not be read: %s — falling through", exc)
529
530 # 2. Identity store lookup (agent key → human key fallback).
531 lookup_url: str | None = remote_url or get_hub_url(repo_root)
532 if lookup_url is None:
533 logger.debug("⚠️ No hub configured — skipping signing identity lookup")
534 return None
535
536 result = resolve_signing_identity(lookup_url, agent_id=agent_id)
537 if result is None:
538 logger.debug(
539 "⚠️ No signing identity for hub %s — run `muse auth keygen && muse auth register`",
540 lookup_url,
541 )
542 return None
543
544 handle, private_key = result
545 logger.debug("✅ Signing identity resolved for hub %s (handle=%s)", lookup_url, handle)
546 return SigningIdentity(handle=handle, private_key=private_key)
547
548 # ---------------------------------------------------------------------------
549 # Hub helpers
550 # ---------------------------------------------------------------------------
551
552 def get_hub_url(repo_root: pathlib.Path | None = None) -> str | None:
553 """Return the hub URL from ``[hub] url``, or ``None`` if not configured.
554
555 Resolution order:
556 1. ``<repo>/.muse/config.toml`` — repo-local config (highest priority).
557 2. ``~/.muse/config.toml`` — global user config (fallback).
558
559 This fallback ensures ``muse auth whoami`` and other hub-aware commands
560 work without ``--hub`` even when invoked outside a repository, as long as
561 the user has set a default hub in their global config.
562
563 Args:
564 repo_root: Repository root. Defaults to ``Path.cwd()``.
565
566 Returns:
567 URL string, or ``None``.
568 """
569 config = _load_config(_config_path(repo_root))
570 hub = config.get("hub")
571 if hub is not None:
572 url = hub.get("url", "")
573 if url.strip():
574 return url.strip()
575
576 # Fall back to ~/.muse/config.toml so hub-aware commands (e.g. `muse auth
577 # whoami`) work without --hub when invoked outside a repository.
578 global_config = _load_config(_GLOBAL_CONFIG_FILE)
579 global_hub = global_config.get("hub")
580 if global_hub is not None:
581 url = global_hub.get("url", "")
582 if url.strip():
583 return url.strip()
584
585 return None
586
587 def set_hub_url(url: str, repo_root: pathlib.Path | None = None) -> None:
588 """Write ``[hub] url`` to ``.muse/config.toml``.
589
590 Preserves all other sections. Creates the config file if absent.
591 Rejects ``http://`` URLs — Muse never contacts a hub over cleartext HTTP.
592
593 Args:
594 url: Hub URL (must be ``https://``).
595 repo_root: Repository root. Defaults to ``Path.cwd()``.
596
597 Raises:
598 ValueError: If *url* does not use the ``https://`` scheme.
599 """
600 _is_loopback = url.startswith("http://localhost") or url.startswith("http://127.0.0.1") or url.startswith("http://[::1]")
601 if not url.startswith("https://") and not _is_loopback:
602 raise ValueError(
603 f"Hub URL must use HTTPS. Got: {url!r}\n"
604 "Muse never connects to a hub over cleartext HTTP.\n"
605 "(Exception: http://localhost and http://127.0.0.1 are allowed for local development.)"
606 )
607 cp = _config_path(repo_root)
608 cp.parent.mkdir(parents=True, exist_ok=True)
609 config = _load_config(cp)
610 config["hub"] = HubConfig(url=url)
611 write_text_atomic(cp, _dump_toml(config))
612 logger.info("✅ Hub URL set to %s", url)
613
614 def clear_hub_url(repo_root: pathlib.Path | None = None) -> None:
615 """Remove the ``[hub]`` section from ``.muse/config.toml``.
616
617 Args:
618 repo_root: Repository root. Defaults to ``Path.cwd()``.
619 """
620 cp = _config_path(repo_root)
621 config = _load_config(cp)
622 if "hub" in config:
623 del config["hub"]
624 write_text_atomic(cp, _dump_toml(config))
625 logger.info("✅ Hub disconnected")
626
627 # ---------------------------------------------------------------------------
628 # Generic dotted-key helpers
629 # ---------------------------------------------------------------------------
630
631 _BlockedNS = dict[str, str]
632 _BLOCKED_NAMESPACES: _BlockedNS = {
633 "auth": "Use `muse auth keygen` and `muse auth register` to manage credentials.",
634 "remotes": "Use `muse remote add/remove/rename` to manage remotes.",
635 "user": "User identity is managed via `muse auth register`. Run `muse auth whoami` to inspect.",
636 }
637
638 _SETTABLE_NAMESPACES = {"hub", "domain", "limits", "commit", "reflog", "symlog", "push"}
639
640 # Default cap values — used when the [limits] section is absent or the key
641 # is not set. These are the same values that were previously hardcoded inside
642 # the individual functions.
643 _DEFAULT_MAX_WALK_COMMITS: int = 10_000
644 _DEFAULT_MAX_ANCESTORS: int = 50_000
645 _DEFAULT_MAX_GRAPH_COMMITS: int = 50_000
646 _DEFAULT_SHARD_PREFIX_LENGTH: int = 2
647
648 def get_limit(key: str, repo_root: pathlib.Path | None = None) -> int:
649 """Return a ``[limits]`` integer cap from config, or its default.
650
651 Args:
652 key: Limit key — one of ``max_walk_commits``, ``max_ancestors``,
653 ``max_graph_commits``.
654 repo_root: Repository root; ``None`` falls back to ``Path.cwd()``.
655
656 Returns:
657 Configured integer value, or the built-in default if not set.
658 """
659 defaults: DefaultsMap = {
660 "max_walk_commits": _DEFAULT_MAX_WALK_COMMITS,
661 "max_ancestors": _DEFAULT_MAX_ANCESTORS,
662 "max_graph_commits": _DEFAULT_MAX_GRAPH_COMMITS,
663 "shard_prefix_length": _DEFAULT_SHARD_PREFIX_LENGTH,
664 }
665 default = defaults.get(key, 10_000)
666 config = _load_config(_config_path(repo_root))
667 limits = config.get("limits") or {}
668 # Explicit key dispatch keeps mypy happy on TypedDict literal-required keys.
669 if key == "max_walk_commits":
670 val: int | None = limits.get("max_walk_commits")
671 elif key == "max_ancestors":
672 val = limits.get("max_ancestors")
673 elif key == "max_graph_commits":
674 val = limits.get("max_graph_commits")
675 elif key == "shard_prefix_length":
676 val = limits.get("shard_prefix_length")
677 else:
678 val = None
679 if isinstance(val, int) and val > 0:
680 return val
681 return default
682
683 def get_config_value(key: str, repo_root: pathlib.Path | None = None) -> str | None:
684 """Get a config value by dotted key (e.g. ``user.handle``, ``hub.url``).
685
686 Returns ``None`` when the key is not set or the namespace is unknown.
687
688 Args:
689 key: Dotted key in ``<namespace>.<subkey>`` form.
690 repo_root: Repository root. Defaults to ``Path.cwd()``.
691
692 Returns:
693 String value, or ``None``.
694 """
695 parts = key.split(".", 1)
696 if len(parts) != 2:
697 return None
698 namespace, subkey = parts
699 config = _load_config(_config_path(repo_root))
700
701 if namespace == "user":
702 # User identity lives in identity.toml, keyed by the configured hub URL.
703 hub_url = (config.get("hub") or {}).get("url", "")
704 if not hub_url:
705 return None
706 try:
707 from muse.core.identity import load_identity, hostname_from_url
708 hostname = hostname_from_url(hub_url)
709 entry = load_identity(hostname)
710 if entry is None:
711 return None
712 if subkey == "handle":
713 return entry.get("handle")
714 if subkey == "type":
715 return entry.get("type")
716 if subkey == "display_name":
717 return entry.get("display_name")
718 if subkey == "email":
719 return entry.get("email")
720 except Exception:
721 pass
722 return None
723
724 if namespace == "hub":
725 hub = config.get("hub") or {}
726 if subkey == "url":
727 return hub.get("url")
728 return None
729
730 if namespace == "domain":
731 domain = config.get("domain") or {}
732 return domain.get(subkey)
733
734 if namespace == "limits":
735 limits = config.get("limits") or {}
736 if subkey == "max_walk_commits":
737 v = limits.get("max_walk_commits")
738 return str(v) if isinstance(v, int) else None
739 if subkey == "max_ancestors":
740 v = limits.get("max_ancestors")
741 return str(v) if isinstance(v, int) else None
742 if subkey == "max_graph_commits":
743 v = limits.get("max_graph_commits")
744 return str(v) if isinstance(v, int) else None
745 if subkey == "shard_prefix_length":
746 v = limits.get("shard_prefix_length")
747 return str(v) if isinstance(v, int) else None
748 return None
749
750 if namespace == "commit":
751 commit = config.get("commit") or {}
752 if subkey == "sign":
753 v = commit.get("sign")
754 if v is True:
755 return "true"
756 if v is False:
757 return "false"
758 return None
759 return None
760
761 if namespace == "reflog":
762 reflog = config.get("reflog") or {}
763 if subkey == "expire-days":
764 v = reflog.get("expire_days")
765 return str(v) if isinstance(v, int) else None
766 return None
767
768 if namespace == "symlog":
769 symlog = config.get("symlog") or {}
770 if subkey == "expire-days":
771 v = symlog.get("expire_days")
772 return str(v) if isinstance(v, int) else None
773 return None
774
775 if namespace == "push":
776 push = config.get("push") or {}
777 if subkey == "tags":
778 v = push.get("tags")
779 if v is True:
780 return "true"
781 if v is False:
782 return "false"
783 return None
784 return None
785
786 return None
787
788 def set_config_value(key: str, value: str, repo_root: pathlib.Path | None = None) -> None:
789 """Set a config value by dotted key (e.g. ``user.handle``, ``domain.ticks_per_beat``).
790
791 Args:
792 key: Dotted key in ``<namespace>.<subkey>`` form.
793 value: New string value.
794 repo_root: Repository root. Defaults to ``Path.cwd()``.
795
796 Raises:
797 ValueError: If the namespace is blocked, unknown, or the subkey is invalid.
798 """
799 parts = key.split(".", 1)
800 if len(parts) != 2:
801 raise ValueError(f"Key must be in 'namespace.subkey' form, got: {key!r}")
802 namespace, subkey = parts
803
804 if namespace in _BLOCKED_NAMESPACES:
805 raise ValueError(_BLOCKED_NAMESPACES[namespace])
806
807 if namespace not in _SETTABLE_NAMESPACES:
808 raise ValueError(
809 f"Unknown config namespace {namespace!r}. "
810 f"Settable namespaces: {', '.join(sorted(_SETTABLE_NAMESPACES))}"
811 )
812
813 cp = _config_path(repo_root)
814 cp.parent.mkdir(parents=True, exist_ok=True)
815 config = _load_config(cp)
816
817 if namespace == "user":
818 set_user_field(subkey, value, repo_root)
819 return
820
821 if namespace == "hub":
822 if subkey != "url":
823 raise ValueError(f"Unknown [hub] config key: {subkey!r}. Valid keys: url")
824 # Route through set_hub_url — it enforces the HTTPS requirement.
825 set_hub_url(value, repo_root)
826 return
827
828 if namespace == "limits":
829 _LIMITS_KEYS = frozenset({
830 "max_walk_commits", "max_ancestors", "max_graph_commits", "shard_prefix_length",
831 })
832 if subkey not in _LIMITS_KEYS:
833 raise ValueError(
834 f"Unknown [limits] config key: {subkey!r}. "
835 f"Valid keys: {', '.join(sorted(_LIMITS_KEYS))}"
836 )
837 try:
838 int_value = int(value)
839 except ValueError as exc:
840 raise ValueError(
841 f"[limits] {subkey} must be an integer, got: {value!r}"
842 ) from exc
843 if int_value <= 0:
844 raise ValueError(f"[limits] {subkey} must be a positive integer, got: {int_value}")
845 if subkey == "shard_prefix_length" and int_value not in (2, 4):
846 raise ValueError("shard_prefix_length must be 2 or 4")
847 limits_section: LimitsConfig = config.get("limits") or {}
848 if subkey == "max_walk_commits":
849 limits_section["max_walk_commits"] = int_value
850 elif subkey == "max_ancestors":
851 limits_section["max_ancestors"] = int_value
852 elif subkey == "max_graph_commits":
853 limits_section["max_graph_commits"] = int_value
854 elif subkey == "shard_prefix_length":
855 limits_section["shard_prefix_length"] = int_value
856 config["limits"] = limits_section
857 write_text_atomic(cp, _dump_toml(config))
858 logger.info("✅ limits.%s = %d", subkey, int_value)
859 return
860
861 if namespace == "commit":
862 _COMMIT_KEYS = frozenset({"sign"})
863 if subkey not in _COMMIT_KEYS:
864 raise ValueError(
865 f"Unknown [commit] config key: {subkey!r}. "
866 f"Valid keys: {', '.join(sorted(_COMMIT_KEYS))}"
867 )
868 if value not in ("true", "false"):
869 raise ValueError(f"[commit] {subkey} must be 'true' or 'false', got: {value!r}")
870 commit_section: CommitConfig = config.get("commit") or {}
871 if subkey == "sign":
872 commit_section["sign"] = value == "true"
873 config["commit"] = commit_section
874 write_text_atomic(cp, _dump_toml(config))
875 logger.info("✅ commit.%s = %s", subkey, value)
876 return
877
878 if namespace == "reflog":
879 _REFLOG_KEYS = frozenset({"expire-days"})
880 if subkey not in _REFLOG_KEYS:
881 raise ValueError(
882 f"Unknown [reflog] config key: {subkey!r}. "
883 f"Valid keys: {', '.join(sorted(_REFLOG_KEYS))}"
884 )
885 if subkey == "expire-days":
886 try:
887 int_value = int(value)
888 except ValueError as exc:
889 raise ValueError(
890 f"[reflog] expire-days must be a positive integer, got: {value!r}"
891 ) from exc
892 if int_value <= 0:
893 raise ValueError(f"[reflog] expire-days must be a positive integer, got: {int_value}")
894 reflog_section: dict = config.get("reflog") or {}
895 reflog_section["expire_days"] = int_value
896 config["reflog"] = reflog_section
897 write_text_atomic(cp, _dump_toml(config))
898 logger.info("✅ reflog.expire-days = %d", int_value)
899 return
900
901 if namespace == "symlog":
902 _SYMLOG_KEYS = frozenset({"expire-days"})
903 if subkey not in _SYMLOG_KEYS:
904 raise ValueError(
905 f"Unknown [symlog] config key: {subkey!r}. "
906 f"Valid keys: {', '.join(sorted(_SYMLOG_KEYS))}"
907 )
908 if subkey == "expire-days":
909 try:
910 sl_int_value = int(value)
911 except ValueError as exc:
912 raise ValueError(
913 f"[symlog] expire-days must be a positive integer, got: {value!r}"
914 ) from exc
915 if sl_int_value <= 0:
916 raise ValueError(f"[symlog] expire-days must be a positive integer, got: {sl_int_value}")
917 symlog_section: SymlogConfig = config.get("symlog") or {}
918 symlog_section["expire_days"] = sl_int_value
919 config["symlog"] = symlog_section
920 write_text_atomic(cp, _dump_toml(config))
921 logger.info("✅ symlog.expire-days = %d", sl_int_value)
922 return
923
924 if namespace == "push":
925 _PUSH_KEYS = frozenset({"tags"})
926 if subkey not in _PUSH_KEYS:
927 raise ValueError(
928 f"Unknown [push] config key: {subkey!r}. "
929 f"Valid keys: {', '.join(sorted(_PUSH_KEYS))}"
930 )
931 if value not in ("true", "false"):
932 raise ValueError(f"[push] {subkey} must be 'true' or 'false', got: {value!r}")
933 push_section: dict = config.get("push") or {}
934 push_section[subkey] = value == "true"
935 config["push"] = push_section
936 write_text_atomic(cp, _dump_toml(config))
937 logger.info("✅ push.%s = %s", subkey, value)
938 return
939
940 # namespace == "domain"
941 _validate_toml_key(subkey, "domain key")
942 domain: DomainConfig = config.get("domain") or {}
943 domain[subkey] = value
944 config["domain"] = domain
945 write_text_atomic(cp, _dump_toml(config))
946 logger.info("✅ domain.%s = %r", subkey, value)
947
948 def config_as_dict(repo_root: pathlib.Path | None = None) -> ConfigTree:
949 """Return the full config as a plain ``dict[str, dict[str, str]]`` for JSON output.
950
951 Credentials are never included — the hub section only contains the URL.
952
953 Args:
954 repo_root: Repository root. Defaults to ``Path.cwd()``.
955
956 Returns:
957 Nested dict suitable for ``json.dumps``.
958 """
959 config = _load_config(_config_path(repo_root))
960 result: ConfigTree = {}
961
962 hub = config.get("hub")
963 if hub:
964 hub_url = hub.get("url", "")
965 if hub_url:
966 result["hub"] = {"url": hub_url}
967
968 remotes = config.get("remotes") or {}
969 if remotes:
970 remotes_dict: ConfigSection = {}
971 for rname, entry in sorted(remotes.items()):
972 url = entry.get("url", "")
973 if url:
974 remotes_dict[rname] = url
975 if remotes_dict:
976 result["remotes"] = remotes_dict
977
978 domain = config.get("domain") or {}
979 if domain:
980 result["domain"] = dict(sorted(domain.items()))
981
982 limits = config.get("limits") or {}
983 if limits:
984 limits_dict: ConfigSection = {}
985 for lk in ("max_walk_commits", "max_ancestors", "max_graph_commits", "shard_prefix_length"):
986 lv = limits.get(lk)
987 if lv is not None:
988 limits_dict[lk] = str(lv)
989 if limits_dict:
990 result["limits"] = limits_dict
991
992 return result
993
994 def config_path_for_editor(repo_root: pathlib.Path | None = None) -> pathlib.Path:
995 """Return the config path for the ``config edit`` command."""
996 return _config_path(repo_root)
997
998 # ---------------------------------------------------------------------------
999 # Branch metadata helpers
1000 # ---------------------------------------------------------------------------
1001
1002 def write_branch_meta(
1003 repo_root: pathlib.Path,
1004 branch_name: str,
1005 *,
1006 intent: str | None = None,
1007 resumable: bool | None = None,
1008 ) -> None:
1009 """Write per-branch metadata to ``[branch."<name>"]`` in ``.muse/config.toml``.
1010
1011 Only the supplied keyword arguments are updated; existing fields
1012 (``remote``, ``merge``, and previously written ``intent``/``resumable``)
1013 are preserved unchanged.
1014
1015 Args:
1016 repo_root: Repository root directory.
1017 branch_name: Name of the branch (e.g. ``"feat/my-thing"``).
1018 intent: Short description of what this branch is for.
1019 resumable: Mark this branch as a resumable agent checkpoint.
1020 """
1021 _validate_toml_key(branch_name, "branch name")
1022 cp = _config_path(repo_root)
1023 cp.parent.mkdir(parents=True, exist_ok=True)
1024 config = _load_config(cp)
1025 branch_map: dict[str, BranchMeta] = dict(config.get("branch") or {})
1026 entry: BranchMeta = dict(branch_map.get(branch_name) or {}) # type: ignore[arg-type]
1027 if intent is not None:
1028 entry["intent"] = intent
1029 if resumable is not None:
1030 entry["resumable"] = resumable
1031 branch_map[branch_name] = entry
1032 config["branch"] = branch_map
1033 write_text_atomic(cp, _dump_toml(config))
1034
1035 def delete_branch_meta(repo_root: pathlib.Path, branch_name: str) -> None:
1036 """Remove the ``[branch."<name>"]`` section from ``.muse/config.toml``.
1037
1038 Called by ``muse branch -d/-D`` after a successful branch deletion so
1039 stale intent/resumable entries do not accumulate indefinitely. No-op
1040 when the branch has no metadata or the config file is absent.
1041 """
1042 cp = _config_path(repo_root)
1043 if not cp.exists():
1044 return
1045 config = _load_config(cp)
1046 branch_map = dict(config.get("branch") or {})
1047 if branch_name not in branch_map:
1048 return
1049 del branch_map[branch_name]
1050 config["branch"] = branch_map
1051 write_text_atomic(cp, _dump_toml(config))
1052
1053 def read_branch_meta(
1054 repo_root: pathlib.Path,
1055 branch_name: str,
1056 ) -> BranchMeta:
1057 """Return per-branch metadata from ``.muse/config.toml``.
1058
1059 Returns an empty dict when the branch has no metadata or the config file
1060 is absent.
1061
1062 Args:
1063 repo_root: Repository root directory.
1064 branch_name: Name of the branch (e.g. ``"feat/my-thing"``).
1065
1066 Returns:
1067 Dict with any of: ``intent`` (str), ``resumable`` (bool),
1068 ``remote`` (str), ``merge`` (str).
1069 """
1070 config = _load_config(_config_path(repo_root))
1071 branch_map = config.get("branch") or {}
1072 return dict(branch_map.get(branch_name) or {})
1073
1074 # ---------------------------------------------------------------------------
1075 # Protected branches helpers
1076 # ---------------------------------------------------------------------------
1077
1078 def get_protected_branches(repo_root: pathlib.Path | None = None) -> list[str]:
1079 """Return the list of protected branch patterns from ``[protected_branches]``.
1080
1081 Returns an empty list when the section is absent or has no ``branches`` key.
1082
1083 Args:
1084 repo_root: Repository root. Defaults to ``Path.cwd()``.
1085 """
1086 config = _load_config(_config_path(repo_root))
1087 return list(config.get("protected_branches") or [])
1088
1089 def is_branch_protected(branch: str, patterns: list[str]) -> bool:
1090 """Return ``True`` if *branch* matches any pattern in *patterns*.
1091
1092 Patterns are matched with :func:`fnmatch.fnmatch` (shell-style globs).
1093 Matching is case-sensitive, consistent with Python fnmatch behaviour.
1094
1095 Args:
1096 branch: Branch name to test (e.g. ``"release/1.0"``).
1097 patterns: List of patterns from ``[protected_branches] branches``.
1098 """
1099 return any(fnmatch.fnmatch(branch, p) for p in patterns)
1100
1101 # ---------------------------------------------------------------------------
1102 # Remote helpers
1103 # ---------------------------------------------------------------------------
1104
1105 def get_remote(name: str, repo_root: pathlib.Path | None = None) -> str | None:
1106 """Return the URL for remote *name*, or ``None`` when not configured.
1107
1108 Args:
1109 name: Remote name (e.g. ``"origin"``).
1110 repo_root: Repository root. Defaults to ``Path.cwd()``.
1111
1112 Returns:
1113 URL string, or ``None``.
1114 """
1115 config = _load_config(_config_path(repo_root))
1116 remotes = config.get("remotes")
1117 if remotes is None:
1118 return None
1119 entry = remotes.get(name)
1120 if entry is None:
1121 return None
1122 url = entry.get("url", "")
1123 return url.strip() if url.strip() else None
1124
1125 def set_remote(
1126 name: str,
1127 url: str,
1128 repo_root: pathlib.Path | None = None,
1129 ) -> None:
1130 """Write ``[remotes.<name>] url`` to ``.muse/config.toml``.
1131
1132 Preserves all other sections. Creates the file if absent.
1133
1134 Args:
1135 name: Remote name (e.g. ``"origin"``).
1136 url: Remote URL.
1137 repo_root: Repository root. Defaults to ``Path.cwd()``.
1138 """
1139 cp = _config_path(repo_root)
1140 cp.parent.mkdir(parents=True, exist_ok=True)
1141 config = _load_config(cp)
1142 existing_remotes = config.get("remotes")
1143 remotes: RemotesMap = {}
1144 if existing_remotes:
1145 remotes.update(existing_remotes)
1146 existing_entry = remotes.get(name)
1147 entry: RemoteEntry = {}
1148 if existing_entry is not None:
1149 if "url" in existing_entry:
1150 entry["url"] = existing_entry["url"]
1151 if "branch" in existing_entry:
1152 entry["branch"] = existing_entry["branch"]
1153 entry["url"] = url
1154 remotes[name] = entry
1155 config["remotes"] = remotes
1156 write_text_atomic(cp, _dump_toml(config))
1157 logger.info("✅ Remote %r set to %s", name, url)
1158
1159 def remove_remote(
1160 name: str,
1161 repo_root: pathlib.Path | None = None,
1162 ) -> None:
1163 """Remove a named remote and its tracking refs.
1164
1165 Args:
1166 name: Remote name to remove.
1167 repo_root: Repository root. Defaults to ``Path.cwd()``.
1168
1169 Raises:
1170 KeyError: If *name* is not a configured remote.
1171 """
1172 cp = _config_path(repo_root)
1173 config = _load_config(cp)
1174 remotes = config.get("remotes")
1175 if remotes is None or name not in remotes:
1176 raise KeyError(name)
1177 del remotes[name]
1178 config["remotes"] = remotes
1179 write_text_atomic(cp, _dump_toml(config))
1180 logger.info("✅ Remote %r removed from config", name)
1181
1182 root = (repo_root or pathlib.Path.cwd()).resolve()
1183 refs_dir = _remote_tracking_dir(root, name)
1184 if refs_dir.is_symlink():
1185 # Refuse to rmtree a symlink — following a symlink placed by an
1186 # attacker could delete files outside the repository tree.
1187 logger.warning("⚠️ Skipping rmtree: remotes dir %s is a symlink", refs_dir)
1188 elif refs_dir.is_dir():
1189 shutil.rmtree(refs_dir)
1190 logger.debug("✅ Removed tracking refs dir %s", refs_dir)
1191
1192 def rename_remote(
1193 old_name: str,
1194 new_name: str,
1195 repo_root: pathlib.Path | None = None,
1196 ) -> None:
1197 """Rename a remote and move its tracking refs.
1198
1199 Args:
1200 old_name: Current remote name.
1201 new_name: Desired new remote name.
1202 repo_root: Repository root. Defaults to ``Path.cwd()``.
1203
1204 Raises:
1205 KeyError: If *old_name* is not a configured remote.
1206 ValueError: If *new_name* is already configured.
1207 """
1208 cp = _config_path(repo_root)
1209 config = _load_config(cp)
1210 remotes = config.get("remotes")
1211 if remotes is None or old_name not in remotes:
1212 raise KeyError(old_name)
1213 if new_name in remotes:
1214 raise ValueError(new_name)
1215 remotes[new_name] = remotes.pop(old_name)
1216 config["remotes"] = remotes
1217 write_text_atomic(cp, _dump_toml(config))
1218 logger.info("✅ Remote %r renamed to %r", old_name, new_name)
1219
1220 root = (repo_root or pathlib.Path.cwd()).resolve()
1221 old_refs_dir = _remote_tracking_dir(root, old_name)
1222 new_refs_dir = _remote_tracking_dir(root, new_name)
1223 if old_refs_dir.is_dir():
1224 old_refs_dir.rename(new_refs_dir)
1225 logger.debug("✅ Moved tracking refs dir %s → %s", old_refs_dir, new_refs_dir)
1226
1227 def list_remotes(repo_root: pathlib.Path | None = None) -> list[RemoteConfig]:
1228 """Return all configured remotes sorted alphabetically by name.
1229
1230 Args:
1231 repo_root: Repository root. Defaults to ``Path.cwd()``.
1232
1233 Returns:
1234 List of ``{"name": str, "url": str}`` dicts.
1235 """
1236 config = _load_config(_config_path(repo_root))
1237 remotes = config.get("remotes")
1238 if remotes is None:
1239 return []
1240 result: list[RemoteConfig] = []
1241 for remote_name in sorted(remotes):
1242 entry = remotes[remote_name]
1243 url = entry.get("url", "")
1244 if not url.strip():
1245 continue
1246 rc = RemoteConfig(name=remote_name, url=url.strip())
1247 result.append(rc)
1248 return result
1249
1250 # ---------------------------------------------------------------------------
1251 # Remote tracking-head helpers
1252 # ---------------------------------------------------------------------------
1253
1254 def _remote_head_path(
1255 remote_name: str,
1256 branch: str,
1257 repo_root: pathlib.Path | None = None,
1258 ) -> pathlib.Path:
1259 """Return the path to the remote tracking pointer file."""
1260 root = (repo_root or pathlib.Path.cwd()).resolve()
1261 return _remote_ref_path(root, remote_name, branch)
1262
1263 def get_remote_head(
1264 remote_name: str,
1265 branch: str,
1266 repo_root: pathlib.Path | None = None,
1267 ) -> str | None:
1268 """Return the last-known remote commit ID for *remote_name*/*branch*.
1269
1270 Returns ``None`` when the tracking pointer does not exist.
1271
1272 Args:
1273 remote_name: Remote name (e.g. ``"origin"``).
1274 branch: Branch name (e.g. ``"main"``).
1275 repo_root: Repository root. Defaults to ``Path.cwd()``.
1276
1277 Returns:
1278 Commit ID string, or ``None``.
1279 """
1280 return read_ref(_remote_head_path(remote_name, branch, repo_root))
1281
1282 def set_remote_head(
1283 remote_name: str,
1284 branch: str,
1285 commit_id: str,
1286 repo_root: pathlib.Path | None = None,
1287 ) -> None:
1288 """Write the remote tracking pointer for *remote_name*/*branch*.
1289
1290 Args:
1291 remote_name: Remote name (e.g. ``"origin"``).
1292 branch: Branch name.
1293 commit_id: Commit ID to record as the known remote HEAD.
1294 repo_root: Repository root. Defaults to ``Path.cwd()``.
1295 """
1296 pointer = _remote_head_path(remote_name, branch, repo_root)
1297 write_text_atomic(pointer, commit_id)
1298 logger.debug("✅ Remote head %s/%s → %s", remote_name, branch, short_id(commit_id))
1299
1300 def delete_remote_head(
1301 remote_name: str,
1302 branch: str,
1303 repo_root: pathlib.Path | None = None,
1304 ) -> bool:
1305 """Remove the local remote-tracking pointer for *remote_name*/*branch*.
1306
1307 Used after ``muse push --delete`` deletes the branch on the server, or when
1308 pruning stale tracking refs with ``muse branch -dr``.
1309
1310 Args:
1311 remote_name: Remote name (e.g. ``"origin"``).
1312 branch: Branch name (e.g. ``"feat/my-thing"``).
1313 repo_root: Repository root. Defaults to ``Path.cwd()``.
1314
1315 Returns:
1316 ``True`` if the pointer file existed and was removed, ``False`` if it
1317 was already absent (idempotent).
1318 """
1319 pointer = _remote_head_path(remote_name, branch, repo_root)
1320 if not pointer.is_file():
1321 return False
1322 pointer.unlink()
1323 # Remove now-empty parent directories (mirrors _cleanup_empty_dirs in branch.py).
1324 remotes_dir = pointer.parent
1325 while remotes_dir.name != remote_name:
1326 try:
1327 remotes_dir.rmdir()
1328 except OSError:
1329 break
1330 remotes_dir = remotes_dir.parent
1331 logger.debug("🗑 Remote tracking ref %s/%s removed", remote_name, branch)
1332 return True
1333
1334 # ---------------------------------------------------------------------------
1335 # Upstream tracking helpers
1336 # ---------------------------------------------------------------------------
1337
1338 def set_upstream(
1339 branch: str,
1340 remote_name: str,
1341 repo_root: pathlib.Path | None = None,
1342 ) -> None:
1343 """Record *remote_name* as the upstream remote for *branch*.
1344
1345 Args:
1346 branch: Local (and remote) branch name.
1347 remote_name: Remote name.
1348 repo_root: Repository root. Defaults to ``Path.cwd()``.
1349 """
1350 cp = _config_path(repo_root)
1351 cp.parent.mkdir(parents=True, exist_ok=True)
1352 config = _load_config(cp)
1353 existing_remotes = config.get("remotes")
1354 remotes: RemotesMap = {}
1355 if existing_remotes:
1356 remotes.update(existing_remotes)
1357 existing_entry = remotes.get(remote_name)
1358 entry: RemoteEntry = {}
1359 if existing_entry is not None:
1360 if "url" in existing_entry:
1361 entry["url"] = existing_entry["url"]
1362 if "branch" in existing_entry:
1363 entry["branch"] = existing_entry["branch"]
1364 entry["branch"] = branch
1365 remotes[remote_name] = entry
1366 config["remotes"] = remotes
1367 write_text_atomic(cp, _dump_toml(config))
1368 logger.info("✅ Upstream for branch %r set to %s/%r", branch, remote_name, branch)
1369
1370 def get_upstream(
1371 branch: str,
1372 repo_root: pathlib.Path | None = None,
1373 ) -> str | None:
1374 """Return the configured upstream remote name for *branch*, or ``None``.
1375
1376 Args:
1377 branch: Local branch name.
1378 repo_root: Repository root. Defaults to ``Path.cwd()``.
1379
1380 Returns:
1381 Remote name string, or ``None``.
1382 """
1383 config = _load_config(_config_path(repo_root))
1384 remotes = config.get("remotes")
1385 if remotes is None:
1386 return None
1387 for rname, entry in remotes.items():
1388 tracked = entry.get("branch", "")
1389 if tracked.strip() == branch:
1390 return rname
1391 return None
1392
1393 # ---------------------------------------------------------------------------
1394 # Global user config — ~/.muse/config.toml (safe_dirs)
1395 # ---------------------------------------------------------------------------
1396
1397 _GLOBAL_MUSE_DIR = _user_muse_dir()
1398 _GLOBAL_CONFIG_FILE = _user_config_toml_path()
1399
1400 def _load_global_config() -> _SecurityConfig:
1401 """Load ``~/.muse/config.toml`` and return the ``[security]`` section.
1402
1403 Returns a dict with key ``safe_dirs`` mapping to a list of path strings.
1404 Returns ``{"safe_dirs": []}`` when the file is absent or unparseable.
1405 """
1406 if not _GLOBAL_CONFIG_FILE.is_file():
1407 return {"safe_dirs": []}
1408 try:
1409 with _GLOBAL_CONFIG_FILE.open("rb") as fh:
1410 raw = tomllib.load(fh)
1411 except Exception as exc: # noqa: BLE001
1412 logger.warning("⚠️ Failed to parse %s: %s", _GLOBAL_CONFIG_FILE, exc)
1413 return {"safe_dirs": []}
1414 security_raw = raw.get("security")
1415 if not isinstance(security_raw, dict):
1416 return {"safe_dirs": []}
1417 dirs_raw = security_raw.get("safe_dirs")
1418 if not isinstance(dirs_raw, list):
1419 return {"safe_dirs": []}
1420 safe: list[str] = [d for d in dirs_raw if isinstance(d, str) and d.strip()]
1421 return {"safe_dirs": safe}
1422
1423 def _save_global_config(safe_dirs: list[str]) -> None:
1424 """Write ``[security] safe_dirs`` to ``~/.muse/config.toml``.
1425
1426 Preserves any other sections that may exist in the file.
1427 """
1428 import os
1429 _GLOBAL_MUSE_DIR.mkdir(parents=True, exist_ok=True)
1430
1431 # Read existing raw content to preserve other sections.
1432 existing_lines: list[str] = []
1433 if _GLOBAL_CONFIG_FILE.is_file():
1434 try:
1435 existing_lines = _GLOBAL_CONFIG_FILE.read_text("utf-8").splitlines()
1436 except Exception: # noqa: BLE001
1437 existing_lines = []
1438
1439 # Strip any existing [security] section from the file.
1440 filtered: list[str] = []
1441 in_security = False
1442 for line in existing_lines:
1443 stripped = line.strip()
1444 if stripped == "[security]":
1445 in_security = True
1446 continue
1447 if in_security and stripped.startswith("["):
1448 in_security = False
1449 if not in_security:
1450 filtered.append(line)
1451
1452 # Remove trailing blank lines before appending the new section.
1453 while filtered and not filtered[-1].strip():
1454 filtered.pop()
1455
1456 # Append the new [security] section.
1457 filtered.append("")
1458 filtered.append("[security]")
1459 if safe_dirs:
1460 items = ", ".join(f'"{_escape(d)}"' for d in safe_dirs)
1461 filtered.append(f"safe_dirs = [{items}]")
1462 else:
1463 filtered.append("safe_dirs = []")
1464 filtered.append("")
1465
1466 content = "\n".join(filtered)
1467 tmp = _GLOBAL_CONFIG_FILE.with_suffix(".toml.tmp")
1468 tmp.write_text(content, encoding="utf-8")
1469 os.replace(tmp, _GLOBAL_CONFIG_FILE)
1470
1471 def get_global_safe_dirs() -> list[str]:
1472 """Return the ``safe_dirs`` list from ``~/.muse/config.toml``.
1473
1474 Returns an empty list when not configured.
1475 """
1476 return _load_global_config().get("safe_dirs", [])
1477
1478 def add_global_safe_dir(path: str) -> None:
1479 """Add *path* to the ``safe_dirs`` list in ``~/.muse/config.toml``.
1480
1481 Normalises the path (``os.path.abspath``) before storing. Idempotent —
1482 adding the same path twice has no effect.
1483
1484 Args:
1485 path: Absolute or relative path to trust.
1486 """
1487 import os
1488 abs_path = os.path.abspath(path)
1489 current = get_global_safe_dirs()
1490 if abs_path not in current:
1491 current.append(abs_path)
1492 _save_global_config(current)
1493 logger.info("✅ Trusted path added: %s", abs_path)
1494
1495 def remove_global_safe_dir(path: str) -> None:
1496 """Remove *path* from the ``safe_dirs`` list in ``~/.muse/config.toml``.
1497
1498 Normalises the path before matching. No-op when the path is not present.
1499
1500 Args:
1501 path: Absolute or relative path to remove from the trust list.
1502 """
1503 import os
1504 abs_path = os.path.abspath(path)
1505 current = get_global_safe_dirs()
1506 updated = [d for d in current if d != abs_path]
1507 _save_global_config(updated)
1508 logger.info("✅ Trusted path removed: %s", abs_path)
File History 10 commits
sha256:c287f599c5429903a139eadf3c5db5d930520e57cb0c3c575d9570e953c3b2d6 chore: bump version to 0.2.0.dev2 — nightly.2 Sonnet 4.6 patch 12 days ago
sha256:8de4334a98c945aace420969d389ad678aa926d4ab4e886b2ac4c4241cb3bf2b revert: keep pyproject.toml in canonical PEP 440 form Sonnet 4.6 patch 15 days ago
sha256:50bb615c573aa4b928fca75b60f82ba2a659910066507cec6a95c412ae22ccb9 docs: add issue docs for push have-negotiation bug (#55) an… Sonnet 4.6 19 days ago
sha256:d90d175cded68aae1d4ffcf4858917854195d0cd8ce1fe73cee4dbc02541cb74 chore: trigger push to surface null-OID paths Sonnet 4.6 24 days ago
sha256:e452ad9a6ace6ccc6d875a35e06caf9da5576a970c1c36133b69a891ce5fefa8 chore: prebuild timing test Sonnet 4.6 34 days ago
sha256:0008ab6695e3e064b3e236b24fd19e538fef6a588eb0d211622f4466d919c0b1 merge: pull staging/dev — advance to 0.2.0rc12 Sonnet 4.6 patch 36 days ago
sha256:9c33d61749fff814c5226d5386aa2af7064c2c02788594a25fdd709358132eea fix: _PROPOSAL_PREFIX_RESOLVE_LIMIT 200 → 100 to match hub … Sonnet 4.6 48 days ago
sha256:36c3cb3e76619d4c30a6d9bf81b5ec4ff148e30dcfed913e3114ca7b43b81c7e fix: rename objects→blobs in push client and all stale test… Sonnet 4.6 patch 50 days ago
sha256:c06a9b9b9fee26c68ea725b44d54b2c0a171301ce9de746d5b656617b4463a9a fix: repair four test failures from post-migration audit Sonnet 4.6 patch 57 days ago
sha256:1900655993c83c4107067375548a7be823e471d2515830842f1a12cba4bd3cdf fix: unified object store migration — idempotent writes, JS… Sonnet 4.6 minor ⚠ 57 days ago