Coverage for src / rhiza / models / _git / lock_io.py: 100%
104 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-07-16 12:51 +0000
1"""Lock-file persistence and orphan-cleanup for the 3-way merge.
3These helpers own reading/writing ``template.lock`` and reconciling the set of
4template-managed files on disk after a merge. They live in the models layer
5because :mod:`rhiza.models._git.merge` drives them as part of the sync
6strategy; keeping them here avoids a ``models -> commands`` import cycle (see
7ADR-0005). The ``rhiza.commands._sync_helpers`` module re-exports them so
8existing command-layer call sites keep working.
9"""
11import contextlib
12import dataclasses
13import os
14from pathlib import Path
15from typing import Any
17try:
18 import fcntl
20 _FCNTL_AVAILABLE = True
21except ImportError: # pragma: no cover - Windows
22 _FCNTL_AVAILABLE = False
24from loguru import logger
26from rhiza.models.lock import TemplateLock
28# ---------------------------------------------------------------------------
29# Lock-file constant
30# ---------------------------------------------------------------------------
32LOCK_FILE = ".rhiza/template.lock"
35# ---------------------------------------------------------------------------
36# Orphan-cleanup helpers
37# ---------------------------------------------------------------------------
40def _warn_about_workflow_files(template_files: list[Path]) -> None:
41 """Warn if the template provides workflow files.
43 Args:
44 template_files: List of synced template file paths.
45 """
46 workflow_files = [p for p in template_files if p.parts[:2] == (".github", "workflows")]
48 if workflow_files:
49 logger.warning(
50 "Workflow files were synced. Updating these files requires "
51 "a token with the 'workflow' permission in GitHub Actions."
52 )
53 logger.info(f"Workflow files affected: {len(workflow_files)}")
56def _files_from_snapshot(snapshot_dir: Path) -> set[Path]:
57 """Return all files in *snapshot_dir* as paths relative to that directory.
59 Args:
60 snapshot_dir: Root of a snapshot directory tree.
62 Returns:
63 Set of relative file paths found under *snapshot_dir*.
64 """
65 return {f.relative_to(snapshot_dir) for f in snapshot_dir.rglob("*") if f.is_file()}
68def _read_previously_tracked_files(
69 target: Path,
70 base_snapshot: Path | None = None,
71 lock_file: Path | None = None,
72) -> set[Path]:
73 """Return the set of files tracked by the last sync.
75 Resolution order:
76 1. ``template.lock.files`` when the field is present and non-empty.
77 2. *base_snapshot* directory listing when provided and non-empty (used as a
78 fallback for lock files that pre-date the ``files`` field).
80 Args:
81 target: Target repository path.
82 base_snapshot: Optional directory containing the template snapshot at
83 the previously-synced SHA. When the lock file has no ``files``
84 entry this snapshot is used to reconstruct the tracked-file list,
85 avoiding an extra network fetch.
86 lock_file: Optional explicit path to the lock file. When ``None`` the
87 default ``<target>/.rhiza/template.lock`` is used.
89 Returns:
90 Set of previously tracked file paths (relative to target), or an empty
91 set when no tracking information is found.
92 """
93 if lock_file is None:
94 lock_file = target / ".rhiza" / "template.lock"
96 from_lock = _files_from_lock(lock_file, base_snapshot)
97 return from_lock if from_lock is not None else set()
100def _files_from_lock(lock_file: Path, base_snapshot: Path | None) -> set[Path] | None:
101 """Return the tracked-file set recorded in *lock_file*, or ``None``.
103 Resolution: the lock's ``files`` field when present, otherwise the
104 *base_snapshot* directory listing when the lock predates that field.
106 Args:
107 lock_file: Path to the ``template.lock`` file.
108 base_snapshot: Optional template snapshot at the previously-synced SHA,
109 used to reconstruct the file list when the lock has no ``files``.
111 Returns:
112 The tracked-file set, or ``None`` when the lock yields no usable list
113 (missing/unreadable lock, or empty with no snapshot fallback).
114 """
115 if not lock_file.exists():
116 return None
117 try:
118 lock = TemplateLock.from_yaml(lock_file)
119 except Exception as e: # noqa: BLE001 # best-effort lock read; any parse/IO error yields no tracked-file list
120 logger.debug(f"Could not read template.lock for orphan cleanup: {e}")
121 return None
123 if lock.files:
124 files = {Path(f) for f in lock.files}
125 logger.debug(f"Reading previous file list from template.lock ({len(files)} files)")
126 return files
127 # Lock exists but has no files list - try to reconstruct from the base
128 # snapshot that was already fetched during this sync run.
129 if base_snapshot is not None and base_snapshot.is_dir():
130 snapshot_files = _files_from_snapshot(base_snapshot)
131 if snapshot_files:
132 logger.debug(f"Reconstructing previous file list from base snapshot ({len(snapshot_files)} files)")
133 return snapshot_files
134 return None
137def _delete_orphaned_file(target: Path, file_path: Path) -> None:
138 """Delete a single orphaned file from the target repository.
140 Args:
141 target: Target repository path.
142 file_path: Relative path of the orphaned file to delete.
143 """
144 full_path = target / file_path
145 if full_path.exists():
146 try:
147 full_path.unlink()
148 logger.success(f"[DEL] {file_path}")
149 except OSError as e:
150 logger.warning(f"Failed to delete {file_path}: {e}")
151 else:
152 logger.debug(f"Skipping {file_path} (already deleted)")
155def _clean_orphaned_files(
156 target: Path,
157 template_files: list[Path],
158 base_snapshot: Path | None = None,
159 excludes: set[str] | None = None,
160 previously_tracked_files: set[Path] | None = None,
161 lock_file: Path | None = None,
162) -> None:
163 """Clean up files that are no longer maintained by template.
165 Files that are explicitly excluded via the ``exclude:`` setting in
166 ``template.yml`` are never deleted even if they appear in a previous lock
167 but are absent from *template_files*.
169 Args:
170 target: Target repository path.
171 template_files: List of files currently provided by the template.
172 base_snapshot: Optional directory containing the template snapshot at
173 the previously-synced SHA. Passed through to
174 :func:`_read_previously_tracked_files` as a fallback when the lock
175 file has no ``files`` entry. Ignored when *previously_tracked_files*
176 is supplied directly.
177 excludes: Optional set of relative path strings that are currently
178 excluded from the template sync. Any previously-tracked file
179 present in this set is kept (the user explicitly opted it out).
180 previously_tracked_files: Optional pre-read set of files that were
181 tracked by the previous sync. When supplied this takes precedence
182 over reading from the on-disk lock file, which allows callers to
183 snapshot the old state before the lock is overwritten by the merge.
184 lock_file: Optional explicit path to the lock file. When ``None`` the
185 default ``<target>/.rhiza/template.lock`` is used.
186 """
187 if previously_tracked_files is None:
188 previously_tracked_files = _read_previously_tracked_files(
189 target, base_snapshot=base_snapshot, lock_file=lock_file
190 )
191 if not previously_tracked_files:
192 return
194 logger.debug(f"Found {len(previously_tracked_files)} file(s) in previous tracking")
196 orphaned_files = previously_tracked_files - set(template_files)
198 # Don't delete files that the user has explicitly excluded — they have
199 # opted those files out of template management and want to keep them.
200 if excludes:
201 excluded_as_paths = {Path(e) for e in excludes}
202 orphaned_files = orphaned_files - excluded_as_paths
204 protected_files = {Path(".rhiza/template.yml")}
206 if not orphaned_files:
207 logger.debug("No orphaned files to clean up")
208 return
210 logger.info(f"Found {len(orphaned_files)} orphaned file(s) no longer maintained by template")
211 for file_path in sorted(orphaned_files):
212 if file_path in protected_files:
213 logger.info(f"Skipping protected file: {file_path}")
214 continue
215 _delete_orphaned_file(target, file_path)
218# ---------------------------------------------------------------------------
219# Lock-file helpers
220# ---------------------------------------------------------------------------
223def _lock_content_unchanged(lock: TemplateLock, lock_path: Path) -> bool:
224 """Return ``True`` when the on-disk lock is identical to *lock*, ignoring ``synced_at``.
226 Compares all fields except ``synced_at`` so that a re-sync that produces no
227 template changes does not cause a spurious timestamp-only update to
228 ``template.lock``.
230 Args:
231 lock: The candidate :class:`~rhiza.models.TemplateLock` to write.
232 lock_path: Path to the existing lock file on disk.
234 Returns:
235 ``True`` if the existing lock has the same effective content as *lock*,
236 ``False`` when the lock file does not exist or differs in any field
237 other than ``synced_at``.
238 """
239 if not lock_path.exists():
240 return False
241 try:
242 existing = TemplateLock.from_yaml(lock_path)
243 except Exception: # noqa: BLE001 # any malformed/unreadable lock means "not equal" — a safe default
244 return False
245 return _lock_identity(existing) == _lock_identity(lock)
248def _lock_identity(lock: TemplateLock) -> tuple[Any, ...]:
249 """Return the content-comparison key for *lock*, excluding ``synced_at``.
251 Args:
252 lock: The lock whose effective content should be summarised.
254 Returns:
255 A tuple of every field that participates in the content comparison, so
256 two locks are content-equal iff their identities are equal.
257 """
258 return (
259 lock.sha,
260 lock.repo,
261 str(lock.host),
262 lock.ref,
263 lock.include,
264 lock.exclude,
265 lock.templates,
266 lock.files,
267 lock.strategy,
268 )
271def _write_lock(target: Path, lock: TemplateLock, lock_file: Path | None = None) -> None:
272 """Persist the lock data to the YAML lock file.
274 Writes to a ``.tmp`` sibling file first, then replaces the real lock file
275 atomically with ``os.replace()``. An exclusive advisory lock (via
276 ``fcntl.flock``) is held for the entire write + rename sequence when
277 ``fcntl`` is available so that concurrent writers do not corrupt the file.
278 Falls back silently on platforms without ``fcntl`` (e.g. Windows).
280 Only files that actually exist in *target* are recorded in ``lock.files``.
281 This guarantees that the lock never references paths that are absent from
282 the repository.
284 The lock file is **not** rewritten when the effective content (all fields
285 except ``synced_at``) is identical to what is already on disk, preventing
286 spurious timestamp-only changes when no template updates are available.
288 Args:
289 target: Path to the target repository.
290 lock: The :class:`~rhiza.models.TemplateLock` to record.
291 lock_file: Optional explicit path for the lock file. When ``None`` the
292 default ``<target>/.rhiza/template.lock`` is used.
293 """
294 # Filter the files list to only include paths that exist on disk so that
295 # the lock never contains entries for files that are absent from the repo.
296 # Always sort the resulting list alphabetically.
297 existing_files = sorted(f for f in lock.files if (target / f).exists())
298 missing = sorted(set(lock.files) - set(existing_files))
299 if missing:
300 missing_str = ", ".join(missing)
301 logger.warning(f"{len(missing)} file(s) in lock absent from target and excluded: {missing_str}")
302 lock = dataclasses.replace(lock, files=existing_files)
304 lock_path = lock_file if lock_file is not None else target / LOCK_FILE
306 # Skip write when only synced_at would change to avoid spurious diffs.
307 if _lock_content_unchanged(lock, lock_path):
308 logger.info(f"{lock_path.name} is already up to date — skipping write")
309 return
311 tmp_path = Path(str(lock_path) + ".tmp")
312 lock_path.parent.mkdir(parents=True, exist_ok=True)
313 # Acquire an exclusive advisory lock via a dedicated lock-fd file so that
314 # the flock survives the os.replace() rename of the actual lock file.
315 lock_fd_path = Path(str(lock_path) + ".fd")
316 try:
317 with lock_fd_path.open("a", encoding="utf-8") as lock_fd:
318 if _FCNTL_AVAILABLE:
319 fcntl.flock(lock_fd, fcntl.LOCK_EX)
320 else:
321 logger.debug("fcntl not available - skipping advisory lock on write")
322 lock.to_yaml(tmp_path)
323 os.replace(tmp_path, lock_path)
324 finally:
325 # Best-effort cleanup of the fd file; failures here are non-critical.
326 with contextlib.suppress(OSError):
327 lock_fd_path.unlink(missing_ok=True)
328 logger.info(f"Updated {lock_path.name} -> {lock.sha[:12]}")