cgcardona / muse public
muse_revert.py python
426 lines 16.4 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Muse Revert Service — create a new commit that undoes a prior commit.
2
3 Revert is the safe undo: given a target commit C with parent P, it creates
4 a new commit whose snapshot is P's snapshot (the state before C was applied).
5 History is preserved — no commit is deleted or rewritten.
6
7 For path-scoped reverts (--track, --section), only paths matching the filter
8 prefix are reverted to P's state; all other paths remain at HEAD's state.
9
10 Boundary rules:
11 - Must NOT import StateStore, EntityRegistry, or get_or_create_store.
12 - Must NOT import executor modules or maestro_* handlers.
13 - May import muse_cli.db, muse_cli.models, muse_cli.merge_engine,
14 muse_cli.snapshot.
15
16 Domain analogy: a producer accidentally committed a bad drum arrangement.
17 ``muse revert <commit>`` creates a new "undo commit" so the DAW history
18 shows what happened and when, rather than silently rewriting the timeline.
19 """
20 from __future__ import annotations
21
22 import datetime
23 import logging
24 import pathlib
25 from dataclasses import dataclass, field
26 from typing import Optional
27
28 from sqlalchemy.ext.asyncio import AsyncSession
29
30 from maestro.muse_cli.db import (
31 get_commit_snapshot_manifest,
32 get_head_snapshot_id,
33 insert_commit,
34 resolve_commit_ref,
35 upsert_snapshot,
36 )
37 from maestro.muse_cli.merge_engine import read_merge_state
38 from maestro.muse_cli.models import MuseCliCommit
39 from maestro.muse_cli.snapshot import compute_commit_id, compute_snapshot_id
40
41 logger = logging.getLogger(__name__)
42
43
44 # ---------------------------------------------------------------------------
45 # Result types
46 # ---------------------------------------------------------------------------
47
48
49 @dataclass(frozen=True)
50 class RevertResult:
51 """Outcome of a ``muse revert`` operation.
52
53 Attributes:
54 commit_id: The new commit ID created by the revert (empty when
55 ``no_commit=True`` or when there was nothing to revert).
56 target_commit_id: The commit that was reverted.
57 parent_commit_id: The parent of the reverted commit (whose snapshot
58 the revert restores).
59 revert_snapshot_id: Snapshot ID of the new reverted state.
60 message: The auto-generated or user-supplied commit message.
61 no_commit: True when the revert was staged but not committed.
62 noop: True when reverting would produce no change.
63 scoped_paths: Paths that were selectively reverted (empty = full revert).
64 paths_deleted: Paths removed from muse-work/ during ``--no-commit``.
65 paths_missing: Paths that could not be restored (no bytes on disk);
66 only populated for ``--no-commit`` runs.
67 branch: Branch on which the revert commit was created.
68 """
69
70 commit_id: str
71 target_commit_id: str
72 parent_commit_id: str
73 revert_snapshot_id: str
74 message: str
75 no_commit: bool
76 noop: bool
77 scoped_paths: tuple[str, ...]
78 paths_deleted: tuple[str, ...]
79 paths_missing: tuple[str, ...]
80 branch: str
81
82
83 # ---------------------------------------------------------------------------
84 # Pure helpers
85 # ---------------------------------------------------------------------------
86
87
88 def _filter_paths(
89 manifest: dict[str, str],
90 track: Optional[str],
91 section: Optional[str],
92 ) -> set[str]:
93 """Return the set of paths in *manifest* that match the given filters.
94
95 A path matches if it starts with ``tracks/<track>/`` (for --track)
96 or ``sections/<section>/`` (for --section). When both are supplied the
97 union of matching paths is returned.
98
99 Returns all paths in *manifest* when neither filter is given.
100 """
101 if not track and not section:
102 return set(manifest.keys())
103
104 matched: set[str] = set()
105 for path in manifest:
106 if track and path.startswith(f"tracks/{track}/"):
107 matched.add(path)
108 if section and path.startswith(f"sections/{section}/"):
109 matched.add(path)
110 return matched
111
112
113 def compute_revert_manifest(
114 *,
115 parent_manifest: dict[str, str],
116 head_manifest: dict[str, str],
117 track: Optional[str] = None,
118 section: Optional[str] = None,
119 ) -> tuple[dict[str, str], tuple[str, ...]]:
120 """Compute the manifest that represents the reverted state.
121
122 For an unscoped revert the result is ``parent_manifest`` verbatim.
123 For a scoped revert (--track or --section) the result is ``head_manifest``
124 with the filtered paths replaced by their values from ``parent_manifest``
125 (or removed if they did not exist in the parent).
126
127 Returns:
128 Tuple of (revert_manifest, scoped_paths_tuple). ``scoped_paths_tuple``
129 is empty for an unscoped revert.
130
131 Pure function — no I/O, no DB.
132 """
133 if not track and not section:
134 return dict(parent_manifest), ()
135
136 # Identify paths affected by the filter across both manifests
137 filter_targets = _filter_paths(parent_manifest, track, section) | _filter_paths(
138 head_manifest, track, section
139 )
140
141 result = dict(head_manifest)
142 for path in filter_targets:
143 if path in parent_manifest:
144 result[path] = parent_manifest[path]
145 else:
146 # Path existed at HEAD but not in parent → remove it
147 result.pop(path, None)
148
149 return result, tuple(sorted(filter_targets))
150
151
152 # ---------------------------------------------------------------------------
153 # Filesystem materialization (--no-commit)
154 # ---------------------------------------------------------------------------
155
156
157 def apply_revert_to_workdir(
158 *,
159 workdir: pathlib.Path,
160 revert_manifest: dict[str, str],
161 current_manifest: dict[str, str],
162 ) -> tuple[list[str], list[str]]:
163 """Update *workdir* to match *revert_manifest* as closely as possible.
164
165 Because the Muse object store does not retain file bytes (only sha256
166 hashes), this function can only:
167
168 1. **Delete** files present in *current_manifest* but absent from
169 *revert_manifest* — these are paths that the reverted commit introduced.
170 2. **Warn** about files present in *revert_manifest* but absent from or
171 changed in *workdir* — these need manual restoration.
172
173 Args:
174 workdir: Absolute path to ``muse-work/``.
175 revert_manifest: The target manifest (parent's or scoped mix).
176 current_manifest: The manifest of *workdir* as it stands now.
177
178 Returns:
179 Tuple of (paths_deleted, paths_missing):
180 - ``paths_deleted``: relative paths successfully removed from *workdir*.
181 - ``paths_missing``: relative paths that should exist in the revert
182 state but whose bytes are unavailable (no object store) — the caller
183 must warn the user and ask for manual intervention.
184 """
185 deleted: list[str] = []
186 missing: list[str] = []
187
188 # Remove paths that should not exist after revert
189 for path in sorted(current_manifest):
190 if path not in revert_manifest:
191 abs_path = workdir / path
192 try:
193 abs_path.unlink()
194 deleted.append(path)
195 logger.info("✅ Removed %s from muse-work/", path)
196 except OSError as exc:
197 logger.warning("⚠️ Could not remove %s: %s", path, exc)
198
199 # Identify paths that need restoration but can't be done automatically
200 for path, expected_oid in sorted(revert_manifest.items()):
201 current_oid = current_manifest.get(path)
202 if current_oid != expected_oid:
203 missing.append(path)
204 logger.warning(
205 "⚠️ Cannot restore %s — file bytes not in object store. "
206 "Restore manually or re-run without --no-commit.",
207 path,
208 )
209
210 return deleted, missing
211
212
213 # ---------------------------------------------------------------------------
214 # Async core
215 # ---------------------------------------------------------------------------
216
217
218 async def _revert_async(
219 *,
220 commit_ref: str,
221 root: pathlib.Path,
222 session: AsyncSession,
223 no_commit: bool = False,
224 track: Optional[str] = None,
225 section: Optional[str] = None,
226 ) -> RevertResult:
227 """Core revert pipeline — resolve, validate, and execute the revert.
228
229 Called by the CLI callback and by tests. All filesystem and DB
230 side-effects are isolated here so tests can inject an in-memory
231 SQLite session and a ``tmp_path`` root.
232
233 Args:
234 commit_ref: Commit ID (full or abbreviated) to revert.
235 root: Repo root (must contain ``.muse/``).
236 session: Async DB session (caller owns commit/rollback lifecycle).
237 no_commit: When ``True``, stage changes to muse-work/ but do not
238 create a new commit record.
239 track: Optional track/instrument path prefix filter.
240 section: Optional section path prefix filter.
241
242 Returns:
243 :class:`RevertResult` describing what happened.
244
245 Raises:
246 ``typer.Exit`` with an appropriate exit code on user-facing errors.
247 """
248 import json
249
250 import typer
251
252 from maestro.muse_cli.errors import ExitCode
253 from maestro.muse_cli.snapshot import build_snapshot_manifest
254
255 muse_dir = root / ".muse"
256
257 # ── Guard: block revert during in-progress merge ─────────────────────
258 merge_state = read_merge_state(root)
259 if merge_state is not None and merge_state.conflict_paths:
260 typer.echo(
261 "❌ Revert blocked: unresolved merge conflicts in progress.\n"
262 " Resolve all conflicts, then run 'muse commit' before reverting."
263 )
264 raise typer.Exit(code=ExitCode.USER_ERROR)
265
266 # ── Repo identity ────────────────────────────────────────────────────
267 repo_data: dict[str, str] = json.loads((muse_dir / "repo.json").read_text())
268 repo_id = repo_data["repo_id"]
269
270 head_ref = (muse_dir / "HEAD").read_text().strip()
271 branch = head_ref.rsplit("/", 1)[-1]
272
273 # ── Resolve target commit ────────────────────────────────────────────
274 target_commit = await resolve_commit_ref(session, repo_id, branch, commit_ref)
275 if target_commit is None:
276 typer.echo(f"❌ Commit not found: {commit_ref!r}")
277 raise typer.Exit(code=ExitCode.USER_ERROR)
278
279 target_commit_id = target_commit.commit_id
280
281 # ── Resolve HEAD commit ───────────────────────────────────────────────
282 head_commit = await resolve_commit_ref(session, repo_id, branch, None)
283 head_snapshot_id = head_commit.snapshot_id if head_commit else None
284
285 # ── Get manifests ────────────────────────────────────────────────────
286 # Parent manifest: the state before the target commit was applied
287 parent_manifest: dict[str, str] = {}
288 parent_commit_id: str = ""
289
290 if target_commit.parent_commit_id:
291 parent_commit_id = target_commit.parent_commit_id
292 parent_snapshot = await get_commit_snapshot_manifest(session, parent_commit_id)
293 if parent_snapshot is not None:
294 parent_manifest = parent_snapshot
295 # If target is the root commit (no parent), reverting it means an empty state
296
297 head_manifest: dict[str, str] = {}
298 if head_snapshot_id and head_commit:
299 from maestro.muse_cli.models import MuseCliSnapshot
300 snap_row = await session.get(MuseCliSnapshot, head_commit.snapshot_id)
301 if snap_row is not None:
302 head_manifest = dict(snap_row.manifest)
303
304 # ── Compute revert manifest ──────────────────────────────────────────
305 revert_manifest, scoped_paths = compute_revert_manifest(
306 parent_manifest=parent_manifest,
307 head_manifest=head_manifest,
308 track=track,
309 section=section,
310 )
311
312 revert_snapshot_id = compute_snapshot_id(revert_manifest)
313
314 # ── Nothing-to-revert guard ──────────────────────────────────────────
315 if head_snapshot_id and revert_snapshot_id == head_snapshot_id:
316 typer.echo("Nothing to revert — working tree already matches the reverted state.")
317 return RevertResult(
318 commit_id="",
319 target_commit_id=target_commit_id,
320 parent_commit_id=parent_commit_id,
321 revert_snapshot_id=revert_snapshot_id,
322 message="",
323 no_commit=no_commit,
324 noop=True,
325 scoped_paths=scoped_paths,
326 paths_deleted=(),
327 paths_missing=(),
328 branch=branch,
329 )
330
331 # ── Auto-generate commit message ─────────────────────────────────────
332 revert_message = f"Revert '{target_commit.message}'"
333
334 # ── --no-commit: apply to working tree only ──────────────────────────
335 if no_commit:
336 workdir = root / "muse-work"
337 current_manifest = build_snapshot_manifest(workdir) if workdir.exists() else {}
338 paths_deleted, paths_missing = apply_revert_to_workdir(
339 workdir=workdir,
340 revert_manifest=revert_manifest,
341 current_manifest=current_manifest,
342 )
343 if paths_missing:
344 typer.echo(
345 "⚠️ Some files cannot be restored automatically (bytes not in object store):\n"
346 + "\n".join(f" missing: {p}" for p in sorted(paths_missing))
347 )
348 if paths_deleted:
349 typer.echo(
350 "✅ Staged revert (--no-commit). Files removed:\n"
351 + "\n".join(f" deleted: {p}" for p in sorted(paths_deleted))
352 )
353 else:
354 typer.echo("⚠️ --no-commit: no file deletions were needed.")
355 return RevertResult(
356 commit_id="",
357 target_commit_id=target_commit_id,
358 parent_commit_id=parent_commit_id,
359 revert_snapshot_id=revert_snapshot_id,
360 message=revert_message,
361 no_commit=True,
362 noop=False,
363 scoped_paths=scoped_paths,
364 paths_deleted=tuple(sorted(paths_deleted)),
365 paths_missing=tuple(sorted(paths_missing)),
366 branch=branch,
367 )
368
369 # ── Persist the revert snapshot (objects already in DB) ──────────────
370 await upsert_snapshot(session, manifest=revert_manifest, snapshot_id=revert_snapshot_id)
371 await session.flush()
372
373 # ── Persist the revert commit ─────────────────────────────────────────
374 head_commit_id = head_commit.commit_id if head_commit else None
375 committed_at = datetime.datetime.now(datetime.timezone.utc)
376 new_commit_id = compute_commit_id(
377 parent_ids=[head_commit_id] if head_commit_id else [],
378 snapshot_id=revert_snapshot_id,
379 message=revert_message,
380 committed_at_iso=committed_at.isoformat(),
381 )
382
383 new_commit = MuseCliCommit(
384 commit_id=new_commit_id,
385 repo_id=repo_id,
386 branch=branch,
387 parent_commit_id=head_commit_id,
388 snapshot_id=revert_snapshot_id,
389 message=revert_message,
390 author="",
391 committed_at=committed_at,
392 )
393 await insert_commit(session, new_commit)
394
395 # ── Update branch HEAD pointer ────────────────────────────────────────
396 ref_path = muse_dir / pathlib.Path(head_ref)
397 ref_path.parent.mkdir(parents=True, exist_ok=True)
398 ref_path.write_text(new_commit_id)
399
400 scope_note = ""
401 if scoped_paths:
402 scope_note = f" (scoped to {len(scoped_paths)} path(s))"
403 typer.echo(
404 f"✅ [{branch} {new_commit_id[:8]}] {revert_message}{scope_note}"
405 )
406 logger.info(
407 "✅ muse revert %s → %s on %r: %s",
408 target_commit_id[:8],
409 new_commit_id[:8],
410 branch,
411 revert_message,
412 )
413
414 return RevertResult(
415 commit_id=new_commit_id,
416 target_commit_id=target_commit_id,
417 parent_commit_id=parent_commit_id,
418 revert_snapshot_id=revert_snapshot_id,
419 message=revert_message,
420 no_commit=False,
421 noop=False,
422 scoped_paths=scoped_paths,
423 paths_deleted=(),
424 paths_missing=(),
425 branch=branch,
426 )