aaronrene / gabriel-muse public
workdir.py python
141 lines 5.9 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 33 days ago
1 """workdir.py — Working-tree restoration utilities.
2
3 The Muse working tree is the repository root (minus ``.muse/`` and other
4 hidden/ignored paths tracked by ``.museignore``). These helpers surgically
5 apply a snapshot manifest to the working tree without ever destroying the
6 root directory itself.
7 """
8
9 from __future__ import annotations
10
11 import hashlib
12 import logging
13 import pathlib
14
15 from muse.core.object_store import restore_object
16 from muse.core.snapshot import walk_workdir
17 from muse.core.validation import contain_path
18 from muse.core._types import Manifest
19
20 logger = logging.getLogger(__name__)
21
22
23 def apply_manifest(root: pathlib.Path, target_manifest: Manifest) -> None:
24 """Surgically apply *target_manifest* to the working tree at *root*.
25
26 Unlike a wipe-and-restore approach this function:
27
28 1. Removes files currently visible to Muse that are absent from
29 *target_manifest*.
30 2. Restores every file listed in *target_manifest* from the object store,
31 overwriting any existing content.
32
33 The repository root and ``.muse/`` are never deleted. Only files that
34 were previously tracked (visible to ``walk_workdir``) are candidates for
35 removal, so untracked files (hidden files, ignored paths) are left alone.
36
37 Args:
38 root: Repository root — the directory that contains ``.muse/``.
39 target_manifest: Mapping of POSIX-relative paths to SHA-256 object IDs
40 that the working tree should contain after this call.
41
42 Raises:
43 ValueError: When *target_manifest* is empty but the working tree has
44 tracked files — this is almost certainly a programming error
45 (e.g. a caller passed an unintentionally empty manifest after
46 a failed store read) and would silently delete all tracked files.
47 Callers that genuinely intend to produce an empty working tree
48 must call ``walk_workdir`` first and handle this case explicitly.
49 """
50 current_files = set(walk_workdir(root).keys())
51 target_files = set(target_manifest.keys())
52
53 # Last-resort data-loss guard: an empty target_manifest with a non-empty
54 # working tree is almost never intentional. The canonical way to produce
55 # this is compute_snapshot_id({}) → SHA-256(b"") — the data-loss sentinel.
56 # Refuse the call so the caller's failure surfaces as an exception rather
57 # than as mysteriously missing files.
58 if not target_manifest and current_files:
59 raise ValueError(
60 f"apply_manifest called with an empty target_manifest while the "
61 f"working tree contains {len(current_files)} tracked file(s). "
62 "This would delete all tracked files and is almost certainly a "
63 "programming error (e.g. an unreadable snapshot was silently treated "
64 "as {}). Fix the root cause in the caller instead."
65 )
66
67 for rel_posix in current_files - target_files:
68 fp = root / pathlib.Path(rel_posix)
69 if fp.exists():
70 fp.unlink()
71
72 missing: list[str] = []
73 for rel_path, object_id in target_manifest.items():
74 try:
75 safe_dest = contain_path(root, rel_path)
76 except ValueError as exc:
77 logger.warning("⚠️ Skipping unsafe manifest path %r: %s", rel_path, exc)
78 continue
79 if not restore_object(root, object_id, safe_dest):
80 missing.append(rel_path)
81
82 if missing:
83 paths_fmt = ", ".join(repr(p) for p in missing[:5])
84 extra = f" … and {len(missing) - 5} more" if len(missing) > 5 else ""
85 raise RuntimeError(
86 f"apply_manifest: {len(missing)} object(s) missing from the local store "
87 f"({paths_fmt}{extra}). The working tree cannot be fully restored. "
88 "Fetch the missing objects from the remote before retrying."
89 )
90
91
92 def verify_workdir_integrity(
93 root: pathlib.Path,
94 expected_manifest: Manifest,
95 ) -> list[tuple[str, str, str | None]]:
96 """Compare the working tree against *expected_manifest* byte-for-byte.
97
98 Reads every file in *expected_manifest* from disk, hashes it, and checks
99 the result against the manifest entry. Also reports tracked files that
100 exist on disk but are not in *expected_manifest*.
101
102 This is an *after-the-fact* integrity check. Call it after
103 :func:`apply_manifest` or after a checkout/merge to confirm the working
104 tree is coherent. The check is O(total file bytes) — use it sparingly
105 on very large repositories.
106
107 Args:
108 root: Repository root.
109 expected_manifest: Mapping of POSIX-relative paths to SHA-256 digests
110 that the working tree must contain.
111
112 Returns:
113 A list of ``(rel_path, expected_hash, actual_hash_or_None)`` tuples
114 for every mismatch. ``actual_hash_or_None`` is ``None`` when the file
115 is absent from disk; it is the on-disk SHA-256 when the file exists
116 but has the wrong content. An empty list means the working tree is
117 perfectly clean.
118 """
119 mismatches: list[tuple[str, str, str | None]] = []
120 actual = walk_workdir(root)
121
122 for rel_path, expected_hash in expected_manifest.items():
123 actual_hash = actual.get(rel_path)
124 if actual_hash is None:
125 fp = root / pathlib.Path(rel_path)
126 if fp.exists():
127 h = hashlib.sha256()
128 with fp.open("rb") as fh:
129 for chunk in iter(lambda: fh.read(65536), b""):
130 h.update(chunk)
131 actual_hash = h.hexdigest()
132 if actual_hash != expected_hash:
133 mismatches.append((rel_path, expected_hash, actual_hash))
134 elif actual_hash != expected_hash:
135 mismatches.append((rel_path, expected_hash, actual_hash))
136
137 for rel_path in actual:
138 if rel_path not in expected_manifest:
139 mismatches.append((rel_path, "", actual[rel_path]))
140
141 return mismatches
File History 2 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 33 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 33 days ago