gabriel / muse public
workdir.py python
153 lines 6.5 KB
Raw
sha256:6ed6b11aceb96bef850842a7aaac54c04843fa9fba5d1200e4bc3845e12d4142 docs: add muse-tag-guide mist (complete idiomatic guide wit… Sonnet 4.6 19 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 import logging
10 import pathlib
11
12 from muse.core.object_store import restore_object
13 from muse.core.snapshot import walk_workdir
14 from muse.core.sparse import filter_manifest_sparse, read_sparse_config
15 from muse.core.validation import contain_path
16 from muse.core.types import Manifest, hash_file
17
18 logger = logging.getLogger(__name__)
19
20 def apply_manifest(
21 root: pathlib.Path,
22 prev_manifest: Manifest,
23 target_manifest: Manifest,
24 ) -> None:
25 """Surgically apply *target_manifest* to the working tree at *root*.
26
27 Unlike a wipe-and-restore approach this function:
28
29 1. Removes files that were in *prev_manifest* but are absent from
30 *target_manifest*.
31 2. Restores every file listed in *target_manifest* from the object store,
32 overwriting any existing content.
33
34 Only files listed in *prev_manifest* are candidates for removal. Untracked
35 files — those never recorded in any commit — are left untouched regardless
36 of what appears on disk. Callers must pass the manifest of the commit that
37 was HEAD before this operation as *prev_manifest*.
38
39 When a sparse-checkout configuration exists at ``.muse/sparse-checkout``
40 the *target_manifest* is filtered to only the paths that match the sparse
41 rules before any working-tree I/O occurs. The full manifest is **not**
42 altered — only the set of files written to disk is restricted.
43
44 Args:
45 root: Repository root — the directory that contains ``.muse/``.
46 prev_manifest: Manifest of the HEAD commit before this operation.
47 Files present here but absent from *target_manifest*
48 will be deleted. Pass ``{}`` when there is no prior
49 commit (e.g. initial clone or first pull).
50 target_manifest: Mapping of POSIX-relative paths to SHA-256 object IDs
51 that the working tree should contain after this call.
52
53 Raises:
54 ValueError: When *target_manifest* is empty but *prev_manifest* has
55 tracked files — this is almost certainly a programming error
56 (e.g. a caller passed an unintentionally empty manifest after
57 a failed store read) and would silently delete all tracked files.
58 Callers that genuinely intend to produce an empty working tree
59 must pass the correct *prev_manifest* and accept the deletion.
60 """
61 # Apply sparse filter when configured — full manifest stays intact in store.
62 sparse_cfg = read_sparse_config(root)
63 if sparse_cfg is not None:
64 target_manifest = filter_manifest_sparse(
65 target_manifest,
66 sparse_cfg.get("patterns", []),
67 mode=sparse_cfg.get("mode", "cone"),
68 )
69
70 current_files = set(prev_manifest.keys())
71 target_files = set(target_manifest.keys())
72
73 # Last-resort data-loss guard: an empty target_manifest with previously
74 # tracked files is almost never intentional.
75 if not target_manifest and current_files:
76 raise ValueError(
77 f"apply_manifest called with an empty target_manifest while "
78 f"prev_manifest contains {len(current_files)} tracked file(s). "
79 "This would delete all tracked files and is almost certainly a "
80 "programming error (e.g. an unreadable snapshot was silently treated "
81 "as {}). Fix the root cause in the caller instead."
82 )
83
84 for rel_posix in current_files - target_files:
85 fp = root / pathlib.Path(rel_posix)
86 if fp.exists():
87 fp.unlink()
88
89 missing: list[str] = []
90 for rel_path, object_id in target_manifest.items():
91 try:
92 safe_dest = contain_path(root, rel_path)
93 except ValueError as exc:
94 logger.warning("⚠️ Skipping unsafe manifest path %r: %s", rel_path, exc)
95 continue
96 if not restore_object(root, object_id, safe_dest):
97 missing.append(rel_path)
98
99 if missing:
100 paths_fmt = ", ".join(repr(p) for p in missing[:5])
101 extra = f" … and {len(missing) - 5} more" if len(missing) > 5 else ""
102 raise RuntimeError(
103 f"apply_manifest: {len(missing)} object(s) missing from the local store "
104 f"({paths_fmt}{extra}). The working tree cannot be fully restored. "
105 "Fetch the missing objects from the remote before retrying."
106 )
107
108 def verify_workdir_integrity(
109 root: pathlib.Path,
110 expected_manifest: Manifest,
111 ) -> list[tuple[str, str, str | None]]:
112 """Compare the working tree against *expected_manifest* byte-for-byte.
113
114 Reads every file in *expected_manifest* from disk, hashes it, and checks
115 the result against the manifest entry. Also reports tracked files that
116 exist on disk but are not in *expected_manifest*.
117
118 This is an *after-the-fact* integrity check. Call it after
119 :func:`apply_manifest` or after a checkout/merge to confirm the working
120 tree is coherent. The check is O(total file bytes) — use it sparingly
121 on very large repositories.
122
123 Args:
124 root: Repository root.
125 expected_manifest: Mapping of POSIX-relative paths to SHA-256 digests
126 that the working tree must contain.
127
128 Returns:
129 A list of ``(rel_path, expected_hash, actual_hash_or_None)`` tuples
130 for every mismatch. ``actual_hash_or_None`` is ``None`` when the file
131 is absent from disk; it is the on-disk SHA-256 when the file exists
132 but has the wrong content. An empty list means the working tree is
133 perfectly clean.
134 """
135 mismatches: list[tuple[str, str, str | None]] = []
136 actual = walk_workdir(root)
137
138 for rel_path, expected_hash in expected_manifest.items():
139 actual_hash = actual.get(rel_path)
140 if actual_hash is None:
141 fp = root / pathlib.Path(rel_path)
142 if fp.exists():
143 actual_hash = hash_file(fp)
144 if actual_hash != expected_hash:
145 mismatches.append((rel_path, expected_hash, actual_hash))
146 elif actual_hash != expected_hash:
147 mismatches.append((rel_path, expected_hash, actual_hash))
148
149 for rel_path in actual:
150 if rel_path not in expected_manifest:
151 mismatches.append((rel_path, "", actual[rel_path]))
152
153 return mismatches
File History 1 commit
sha256:6ed6b11aceb96bef850842a7aaac54c04843fa9fba5d1200e4bc3845e12d4142 docs: add muse-tag-guide mist (complete idiomatic guide wit… Sonnet 4.6 19 days ago