cgcardona / muse public
config.py python
463 lines 15.8 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Muse CLI configuration helpers.
2
3 Reads and writes ``.muse/config.toml`` — the local repository configuration file.
4
5 The config file supports:
6 - ``[auth] token`` — bearer token for Muse Hub authentication (NEVER logged).
7 - ``[remotes.<name>] url`` — remote Hub URL for push/pull sync.
8
9 Token lifecycle (MVP):
10 1. User obtains a token via ``POST /auth/token``.
11 2. User stores it in ``.muse/config.toml`` under ``[auth] token = "..."``
12 3. CLI commands that contact the Hub read the token here automatically.
13
14 Security note: ``.muse/config.toml`` should be added to ``.gitignore`` to
15 prevent the token from being committed to version control.
16 """
17 from __future__ import annotations
18
19 import logging
20 import pathlib
21 import shutil
22 import tomllib
23 from typing import TypedDict
24
25 logger = logging.getLogger(__name__)
26
27 _CONFIG_FILENAME = "config.toml"
28 _MUSE_DIR = ".muse"
29
30
31 # ---------------------------------------------------------------------------
32 # TypedDicts for structured config data
33 # ---------------------------------------------------------------------------
34
35
36 class RemoteConfig(TypedDict):
37 """A single configured remote — name and URL."""
38
39 name: str
40 url: str
41
42
43 # ---------------------------------------------------------------------------
44 # Internal helpers
45 # ---------------------------------------------------------------------------
46
47
48 def _config_path(repo_root: pathlib.Path | None) -> pathlib.Path:
49 """Return the path to .muse/config.toml for the given (or cwd) root."""
50 root = (repo_root or pathlib.Path.cwd()).resolve()
51 return root / _MUSE_DIR / _CONFIG_FILENAME
52
53
54 def _load_config(config_path: pathlib.Path) -> dict[str, object]:
55 """Load and parse config.toml; return empty dict if absent or unreadable."""
56 if not config_path.is_file():
57 return {}
58 try:
59 with config_path.open("rb") as fh:
60 return tomllib.load(fh)
61 except Exception as exc: # noqa: BLE001
62 logger.warning("⚠️ Failed to parse %s: %s", config_path, exc)
63 return {}
64
65
66 def _dump_toml(data: dict[str, object]) -> str:
67 """Serialize a two-level TOML dict (tables of string values) to text.
68
69 Handles the subset of TOML used by .muse/config.toml:
70 - Top-level tables (``[section]``) whose values are strings.
71 - Nested tables (``[section.subsection]``) whose values are strings.
72
73 Values of other types are coerced to strings and stored as TOML strings.
74 The ``[auth]`` section is always written first so the file is stable.
75 """
76 lines: list[str] = []
77
78 def _write_table(heading: str, mapping: dict[str, object]) -> None:
79 lines.append(f"[{heading}]")
80 for key, val in mapping.items():
81 if isinstance(val, str):
82 escaped = val.replace("\\", "\\\\").replace('"', '\\"')
83 lines.append(f'{key} = "{escaped}"')
84 else:
85 lines.append(f"{key} = {val!r}")
86 lines.append("")
87
88 # Auth section first (stable ordering)
89 if "auth" in data:
90 auth_section = data["auth"]
91 if isinstance(auth_section, dict):
92 _write_table("auth", auth_section)
93
94 # Remotes (nested tables: [remotes.<name>])
95 if "remotes" in data:
96 remotes_section = data["remotes"]
97 if isinstance(remotes_section, dict):
98 for remote_name in sorted(remotes_section):
99 remote_cfg = remotes_section[remote_name]
100 if isinstance(remote_cfg, dict):
101 _write_table(f"remotes.{remote_name}", remote_cfg)
102
103 # Any other top-level sections
104 for key, val in data.items():
105 if key in ("auth", "remotes"):
106 continue
107 if isinstance(val, dict):
108 _write_table(key, val)
109
110 return "\n".join(lines)
111
112
113 # ---------------------------------------------------------------------------
114 # Auth helpers
115 # ---------------------------------------------------------------------------
116
117
118 def get_auth_token(repo_root: pathlib.Path | None = None) -> str | None:
119 """Read ``[auth] token`` from ``.muse/config.toml``.
120
121 Returns the token string if present and non-empty, or ``None`` if the
122 file does not exist, ``[auth]`` is absent, or ``token`` is empty/missing.
123
124 The token value is NEVER logged — log lines mask it as ``"Bearer ***"``.
125
126 Args:
127 repo_root: Explicit repository root. Defaults to the current working
128 directory. In tests, pass a ``tmp_path`` fixture value.
129
130 Returns:
131 The raw token string, or ``None``.
132 """
133 config_path = _config_path(repo_root)
134
135 if not config_path.is_file():
136 logger.debug("⚠️ No %s found at %s", _CONFIG_FILENAME, config_path)
137 return None
138
139 data = _load_config(config_path)
140 auth_section = data.get("auth", {})
141 token: object = auth_section.get("token", "") if isinstance(auth_section, dict) else ""
142 if not isinstance(token, str) or not token.strip():
143 logger.debug("⚠️ [auth] token missing or empty in %s", config_path)
144 return None
145
146 logger.debug("✅ Auth token loaded from %s (Bearer ***)", config_path)
147 return token.strip()
148
149
150 # ---------------------------------------------------------------------------
151 # Remote helpers
152 # ---------------------------------------------------------------------------
153
154
155 def get_remote(name: str, repo_root: pathlib.Path | None = None) -> str | None:
156 """Return the URL for remote *name* from ``[remotes.<name>] url``.
157
158 Returns ``None`` when the config file is absent or the named remote has
159 not been configured. Never raises — callers decide what to do on miss.
160
161 Args:
162 name: Remote name (e.g. ``"origin"``).
163 repo_root: Repository root. Defaults to ``Path.cwd()``.
164
165 Returns:
166 URL string, or ``None``.
167 """
168 config_path = _config_path(repo_root)
169 data = _load_config(config_path)
170 remotes_section = data.get("remotes", {})
171 if not isinstance(remotes_section, dict):
172 return None
173 remote_cfg = remotes_section.get(name, {})
174 if not isinstance(remote_cfg, dict):
175 return None
176 url: object = remote_cfg.get("url", "")
177 if not isinstance(url, str) or not url.strip():
178 return None
179 return url.strip()
180
181
182 def set_remote(
183 name: str,
184 url: str,
185 repo_root: pathlib.Path | None = None,
186 ) -> None:
187 """Write ``[remotes.<name>] url = "<url>"`` to ``.muse/config.toml``.
188
189 Preserves all other sections already in the config file. Creates the
190 ``.muse/`` directory and ``config.toml`` if they do not exist.
191
192 Args:
193 name: Remote name (e.g. ``"origin"``).
194 url: Remote URL (e.g. ``"https://hub.example.com/musehub/repos/my-repo"``).
195 repo_root: Repository root. Defaults to ``Path.cwd()``.
196 """
197 config_path = _config_path(repo_root)
198 config_path.parent.mkdir(parents=True, exist_ok=True)
199
200 data = _load_config(config_path)
201
202 # Ensure the nested structure exists
203 if "remotes" not in data or not isinstance(data["remotes"], dict):
204 data["remotes"] = {}
205 remotes: dict[str, object] = data["remotes"] # type: ignore[assignment]
206 if name not in remotes or not isinstance(remotes[name], dict):
207 remotes[name] = {}
208 remote_entry: dict[str, object] = remotes[name] # type: ignore[assignment]
209 remote_entry["url"] = url
210
211 config_path.write_text(_dump_toml(data), encoding="utf-8")
212 logger.info("✅ Remote %r set to %s", name, url)
213
214
215 def remove_remote(
216 name: str,
217 repo_root: pathlib.Path | None = None,
218 ) -> None:
219 """Remove a named remote and all its tracking refs from ``.muse/``.
220
221 Deletes ``[remotes.<name>]`` from ``config.toml`` and removes the entire
222 ``.muse/remotes/<name>/`` directory tree (tracking head files). Raises
223 ``KeyError`` when the remote does not exist so callers can surface a clear
224 error message to the user.
225
226 Args:
227 name: Remote name to remove (e.g. ``"origin"``).
228 repo_root: Repository root. Defaults to ``Path.cwd()``.
229
230 Raises:
231 KeyError: If *name* is not a configured remote.
232 """
233 config_path = _config_path(repo_root)
234 data = _load_config(config_path)
235
236 remotes_section = data.get("remotes", {})
237 if not isinstance(remotes_section, dict) or name not in remotes_section:
238 raise KeyError(name)
239
240 del remotes_section[name]
241 data["remotes"] = remotes_section
242
243 config_path.write_text(_dump_toml(data), encoding="utf-8")
244 logger.info("✅ Remote %r removed from config", name)
245
246 # Remove tracking refs directory if it exists
247 root = (repo_root or pathlib.Path.cwd()).resolve()
248 refs_dir = root / _MUSE_DIR / "remotes" / name
249 if refs_dir.is_dir():
250 shutil.rmtree(refs_dir)
251 logger.debug("✅ Removed tracking refs dir %s", refs_dir)
252
253
254 def rename_remote(
255 old_name: str,
256 new_name: str,
257 repo_root: pathlib.Path | None = None,
258 ) -> None:
259 """Rename a remote in ``.muse/config.toml`` and move its tracking refs.
260
261 Updates ``[remotes.<old_name>]`` → ``[remotes.<new_name>]`` in config and
262 moves ``.muse/remotes/<old_name>/`` → ``.muse/remotes/<new_name>/``.
263 Raises ``KeyError`` when *old_name* does not exist. Raises ``ValueError``
264 when *new_name* is already configured.
265
266 Args:
267 old_name: Current remote name.
268 new_name: Desired new remote name.
269 repo_root: Repository root. Defaults to ``Path.cwd()``.
270
271 Raises:
272 KeyError: If *old_name* is not a configured remote.
273 ValueError: If *new_name* already exists as a remote.
274 """
275 config_path = _config_path(repo_root)
276 data = _load_config(config_path)
277
278 remotes_section = data.get("remotes", {})
279 if not isinstance(remotes_section, dict) or old_name not in remotes_section:
280 raise KeyError(old_name)
281 if new_name in remotes_section:
282 raise ValueError(new_name)
283
284 remotes_section[new_name] = remotes_section.pop(old_name)
285 data["remotes"] = remotes_section
286
287 config_path.write_text(_dump_toml(data), encoding="utf-8")
288 logger.info("✅ Remote %r renamed to %r", old_name, new_name)
289
290 # Move tracking refs directory if it exists
291 root = (repo_root or pathlib.Path.cwd()).resolve()
292 old_refs_dir = root / _MUSE_DIR / "remotes" / old_name
293 new_refs_dir = root / _MUSE_DIR / "remotes" / new_name
294 if old_refs_dir.is_dir():
295 old_refs_dir.rename(new_refs_dir)
296 logger.debug("✅ Moved tracking refs dir %s → %s", old_refs_dir, new_refs_dir)
297
298
299 def list_remotes(repo_root: pathlib.Path | None = None) -> list[RemoteConfig]:
300 """Return all configured remotes as :class:`RemoteConfig` dicts.
301
302 Returns an empty list when the config file is absent or contains no
303 ``[remotes.*]`` sections. Sorted alphabetically by remote name.
304
305 Args:
306 repo_root: Repository root. Defaults to ``Path.cwd()``.
307
308 Returns:
309 List of ``{"name": str, "url": str}`` dicts.
310 """
311 config_path = _config_path(repo_root)
312 data = _load_config(config_path)
313 remotes_section = data.get("remotes", {})
314 if not isinstance(remotes_section, dict):
315 return []
316
317 result: list[RemoteConfig] = []
318 for remote_name in sorted(remotes_section):
319 cfg = remotes_section[remote_name]
320 if not isinstance(cfg, dict):
321 continue
322 url_val: object = cfg.get("url", "")
323 if isinstance(url_val, str) and url_val.strip():
324 result.append(RemoteConfig(name=remote_name, url=url_val.strip()))
325
326 return result
327
328
329 # ---------------------------------------------------------------------------
330 # Remote tracking-head helpers
331 # ---------------------------------------------------------------------------
332
333
334 def _remote_head_path(
335 remote_name: str,
336 branch: str,
337 repo_root: pathlib.Path | None = None,
338 ) -> pathlib.Path:
339 """Return the path to the remote tracking pointer file.
340
341 The file lives at ``.muse/remotes/<remote_name>/<branch>`` and contains
342 the last known commit_id on that remote branch.
343 """
344 root = (repo_root or pathlib.Path.cwd()).resolve()
345 return root / _MUSE_DIR / "remotes" / remote_name / branch
346
347
348 def get_remote_head(
349 remote_name: str,
350 branch: str,
351 repo_root: pathlib.Path | None = None,
352 ) -> str | None:
353 """Return the last-known remote commit ID for *remote_name*/*branch*.
354
355 Returns ``None`` when the tracking pointer file does not exist (i.e. this
356 branch has never been pushed/pulled).
357
358 Args:
359 remote_name: Remote name (e.g. ``"origin"``).
360 branch: Branch name (e.g. ``"main"``).
361 repo_root: Repository root. Defaults to ``Path.cwd()``.
362
363 Returns:
364 Commit ID string, or ``None``.
365 """
366 pointer = _remote_head_path(remote_name, branch, repo_root)
367 if not pointer.is_file():
368 return None
369 raw = pointer.read_text(encoding="utf-8").strip()
370 return raw if raw else None
371
372
373 def set_remote_head(
374 remote_name: str,
375 branch: str,
376 commit_id: str,
377 repo_root: pathlib.Path | None = None,
378 ) -> None:
379 """Write the remote tracking pointer for *remote_name*/*branch*.
380
381 Creates the ``.muse/remotes/<remote_name>/`` directory if needed.
382
383 Args:
384 remote_name: Remote name (e.g. ``"origin"``).
385 branch: Branch name (e.g. ``"main"``).
386 commit_id: Commit ID to record as the known remote HEAD.
387 repo_root: Repository root. Defaults to ``Path.cwd()``.
388 """
389 pointer = _remote_head_path(remote_name, branch, repo_root)
390 pointer.parent.mkdir(parents=True, exist_ok=True)
391 pointer.write_text(commit_id, encoding="utf-8")
392 logger.debug("✅ Remote head %s/%s → %s", remote_name, branch, commit_id[:8])
393
394
395 # ---------------------------------------------------------------------------
396 # Upstream tracking helpers
397 # ---------------------------------------------------------------------------
398
399
400 def set_upstream(
401 branch: str,
402 remote_name: str,
403 repo_root: pathlib.Path | None = None,
404 ) -> None:
405 """Record *remote_name* as the upstream remote for *branch*.
406
407 Writes ``branch = "<branch>"`` under ``[remotes.<remote_name>]`` in
408 ``.muse/config.toml``. This mirrors the git ``--set-upstream`` behaviour:
409 the local branch knows which remote branch to track for future push/pull.
410
411 Args:
412 branch: Local (and remote) branch name (e.g. ``"main"``).
413 remote_name: Remote name (e.g. ``"origin"``).
414 repo_root: Repository root. Defaults to ``Path.cwd()``.
415 """
416 config_path = _config_path(repo_root)
417 config_path.parent.mkdir(parents=True, exist_ok=True)
418
419 data = _load_config(config_path)
420
421 if "remotes" not in data or not isinstance(data["remotes"], dict):
422 data["remotes"] = {}
423 remotes: dict[str, object] = data["remotes"] # type: ignore[assignment]
424 if remote_name not in remotes or not isinstance(remotes[remote_name], dict):
425 remotes[remote_name] = {}
426 remote_entry: dict[str, object] = remotes[remote_name] # type: ignore[assignment]
427 remote_entry["branch"] = branch
428
429 config_path.write_text(_dump_toml(data), encoding="utf-8")
430 logger.info("✅ Upstream for branch %r set to %s/%r", branch, remote_name, branch)
431
432
433 def get_upstream(
434 branch: str,
435 repo_root: pathlib.Path | None = None,
436 ) -> str | None:
437 """Return the configured upstream remote name for *branch*, or ``None``.
438
439 Reads ``branch`` under every ``[remotes.*]`` section and returns the first
440 remote whose ``branch`` value matches *branch*.
441
442 Args:
443 branch: Local branch name (e.g. ``"main"``).
444 repo_root: Repository root. Defaults to ``Path.cwd()``.
445
446 Returns:
447 Remote name string (e.g. ``"origin"``), or ``None`` when no upstream
448 is configured for *branch*.
449 """
450 config_path = _config_path(repo_root)
451 data = _load_config(config_path)
452 remotes_section = data.get("remotes", {})
453 if not isinstance(remotes_section, dict):
454 return None
455
456 for rname, remote_cfg in remotes_section.items():
457 if not isinstance(remote_cfg, dict):
458 continue
459 tracked_branch: object = remote_cfg.get("branch", "")
460 if isinstance(tracked_branch, str) and tracked_branch.strip() == branch:
461 return str(rname)
462
463 return None