"""Pure filesystem snapshot logic for ``muse commit``. All functions here are side-effect-free (no DB, no I/O besides reading files under ``workdir``). They are kept separate so they can be unit-tested without a database. ID derivation contract (deterministic, no random/UUID components): object_id = sha256(file_bytes).hexdigest() snapshot_id = sha256( NUL.join(sorted(f"{path}NUL{oid}" for path, oid in manifest.items())) ).hexdigest() commit_id = sha256( NUL.join([NUL.join(sorted(parent_ids)), snapshot_id, message, committed_at_iso]) ).hexdigest() The null byte (\\x00) is used as the field separator because it is: - Illegal in POSIX filenames (preventing separator-injection attacks from crafted file paths). - Absent from SHA-256 hex strings (preventing injection via object IDs). - Absent from ISO-8601 timestamps and typical message text. This replaces the previous ``|`` / ``:`` separator scheme which allowed two distinct manifests or commit inputs to produce the same hash if filenames contained those characters. Symlinks in the working tree are excluded from snapshots. Following a symlink that points outside state/ would silently commit the contents of arbitrary filesystem paths. Exclusion policy ---------------- Dotfiles and dot-directories are **tracked by default** — ``.cursorrules``, ``.editorconfig``, ``.eslintrc`` are intentional project configuration that collaborators need. Exclusion is driven entirely by ``.museignore`` plus the built-in secrets blocklist below. The only hard-coded directory skip is ``.muse/`` itself (internal VCS storage) and a performance-only list of directories that are universally noise (``node_modules/``, ``__pycache__/``, ``.venv/`` etc.). """ from __future__ import annotations import fnmatch import hashlib import json import os import pathlib import re import stat as _stat from muse.core._types import Manifest from muse.core.ignore import is_ignored, load_ignore_config, resolve_patterns from muse.core.stat_cache import load_cache # Directories that are always pruned before os.walk descends into them. # These are either internal VCS storage (.muse) or universally-noisy # directories whose contents are never meaningful project source. # Kept as a frozenset for O(1) lookup inside the hot walk loop. _ALWAYS_PRUNE_DIRS: frozenset[str] = frozenset( { ".muse", ".git", "node_modules", "__pycache__", ".venv", "venv", ".tox", ".nox", ".mypy_cache", ".ruff_cache", ".pytest_cache", ".coverage", "htmlcov", "dist", "build", } ) # Built-in secrets blocklist — applied even when .museignore is absent. # This is the last line of defence: these files must never appear in a # snapshot regardless of what a user configures in .museignore. # # Note: .env.example is intentionally NOT listed here — it is the universal # convention for a safe, credential-free environment template and must be # trackable. We block the real secret files explicitly instead of using a # wildcard that would accidentally catch .env.example. _BUILTIN_SECRET_PATTERNS: list[str] = [ ".env", ".env.local", ".env.development", ".env.staging", ".env.production", ".env.prod", ".envrc", "*.pem", "*.key", "*.p12", "*.pfx", ".DS_Store", "Thumbs.db", ] def _build_filename_filter(patterns: list[str]) -> re.Pattern[str] | None: """Compile a combined regex for fast per-file ignore pre-rejection. Translates every *simple* pattern (no ``/`` in the body) into a single alternating regex so ``re.search(fname)`` can reject the overwhelming majority of files in one call instead of N ``fnmatch`` calls. Only patterns without ``/`` in their body are included — they test the filename component. Patterns with an embedded ``/`` (e.g. ``docs/*.md``) or a trailing ``/`` (directory patterns) must still go through the full :func:`~muse.core.ignore.is_ignored` path. Returns ``None`` when *patterns* is empty or contains no simple patterns. Performance: at 75 000 files with 9 builtin patterns, replacing 9 × N ``fnmatch`` calls with one ``re.search`` call reduces ignore-matching overhead by ~10×, dropping warm ``walk_workdir`` time from ~850 ms to ~85 ms at 75 k scale, making the 1-file-change target of < 200 ms achievable. """ translated: list[str] = [] for raw_pat in patterns: body = raw_pat.lstrip("!") # strip negation marker if body.endswith("/"): body = body.rstrip("/") if "/" in body: continue # path-level pattern — needs full is_ignored evaluation translated.append(fnmatch.translate(body)) if not translated: return None return re.compile("(?:" + "|".join(translated) + ")") def load_ignore_patterns(workdir: pathlib.Path) -> list[str]: """Return the combined ignore pattern list for *workdir*. Reads ``.museignore`` from *workdir* and detects the active domain from ``.muse/repo.json``. Falls back to ``"code"`` when either file is absent. The built-in secrets blocklist is always prepended so it cannot be overridden by user configuration. This function is intentionally public so that commands outside ``snapshot.py`` (e.g. ``stash``) can apply the same ignore rules without duplicating the domain-detection logic. """ domain = "code" repo_json = workdir / ".muse" / "repo.json" if repo_json.exists(): try: raw = json.loads(repo_json.read_text(encoding="utf-8")) if isinstance(raw, dict) and isinstance(raw.get("domain"), str): domain = raw["domain"] except (OSError, json.JSONDecodeError): pass config = load_ignore_config(workdir) user_patterns = resolve_patterns(config, domain) return _BUILTIN_SECRET_PATTERNS + user_patterns _SEP = "\x00" def hash_file(path: pathlib.Path) -> str: """Return the sha256 hex digest of a file's raw bytes. This is the ``object_id`` for the given file. Reading in chunks keeps memory usage constant regardless of file size. """ h = hashlib.sha256() with path.open("rb") as fh: for chunk in iter(lambda: fh.read(65536), b""): h.update(chunk) return h.hexdigest() def build_snapshot_manifest(workdir: pathlib.Path) -> Manifest: """Return ``{rel_path: object_id}`` for every tracked file in *workdir*. Preferred public name; delegates to :func:`walk_workdir`. """ return walk_workdir(workdir) def directories_from_manifest(files: Manifest) -> list[str]: """Derive all implicit parent directories from a file manifest. For every file path in *files*, all ancestor directory components are collected. The result is a sorted, deduplicated list of POSIX directory paths relative to the repository root. Empty directories that have no files are not present in *files* and therefore cannot be derived here — they require an explicit ``.musekeep`` marker file so the filesystem walk in :func:`walk_workdir_with_dirs` can detect them. This helper is used by merge / rebase / cherry-pick operations that compute a merged file manifest without performing a fresh filesystem walk, so that every ``SnapshotRecord`` stores a consistent directory list. """ dirs: set[str] = set() for path in files: parts = path.split("/") for i in range(1, len(parts)): dirs.add("/".join(parts[:i])) return sorted(dirs) def walk_workdir_with_dirs( workdir: pathlib.Path, ) -> tuple[dict[str, str], list[str]]: """Walk *workdir* and return ``(files_manifest, sorted_directories)``. A single ``os.walk`` pass collects both the file content map and the list of every non-root directory encountered (minus always-pruned dirs). This is the canonical entry point for commit and status operations that need first-class directory identity. See :func:`walk_workdir` for the exclusion rules that apply to files. Directories follow the same pruning rules — any directory whose name is in :data:`_ALWAYS_PRUNE_DIRS` is never descended into and therefore never appears in the returned list. """ ignore_patterns = load_ignore_patterns(workdir) cache = load_cache(workdir) manifest: Manifest = {} dirs: list[str] = [] root_str = str(workdir) prefix_len = len(root_str) + 1 _filename_filter: re.Pattern[str] | None = _build_filename_filter(ignore_patterns) _has_complex_patterns: bool = any( "/" in p.lstrip("!") for p in ignore_patterns ) for dirpath, dirnames, filenames in os.walk(root_str, followlinks=False): # Prune always-excluded names and any subdirectory that is itself a # nested muse repo (contains .muse/). Nested repos are independent # version-controlled units — their contents belong to their own # snapshot, not the parent repo's. dirnames[:] = [ d for d in dirnames if d not in _ALWAYS_PRUNE_DIRS and not os.path.isdir(os.path.join(dirpath, d, ".muse")) ] # Track every non-root directory we descend into. if dirpath != root_str: rel_dir = dirpath[prefix_len:] if os.sep != "/": rel_dir = rel_dir.replace(os.sep, "/") dirs.append(rel_dir) for fname in filenames: abs_str = os.path.join(dirpath, fname) try: st = os.lstat(abs_str) except OSError: continue if not _stat.S_ISREG(st.st_mode): continue rel = abs_str[prefix_len:] if os.sep != "/": rel = rel.replace(os.sep, "/") if ( _filename_filter is not None and not _filename_filter.search(fname) and not _has_complex_patterns ): manifest[rel] = cache.get_cached(rel, abs_str, st.st_mtime, st.st_size, st.st_ino) continue if is_ignored(rel, ignore_patterns): continue manifest[rel] = cache.get_cached(rel, abs_str, st.st_mtime, st.st_size, st.st_ino) cache.prune(set(manifest)) cache.save() return manifest, sorted(dirs) def walk_workdir(workdir: pathlib.Path) -> Manifest: """Walk *workdir* and return only the file manifest. Thin wrapper around :func:`walk_workdir_with_dirs` that discards the directory list. Callers that need both files and directories should call :func:`walk_workdir_with_dirs` directly to avoid a second filesystem walk. Walk *workdir* recursively and return ``{rel_path: object_id}``. Exclusions (all silent, no warning emitted): - Symlinks — following them could commit content from outside the repo. - Non-regular files — only regular files are included. - Paths matched by ``.museignore`` or the built-in secrets blocklist. - Directories in ``_ALWAYS_PRUNE_DIRS`` — internal VCS storage and universally-noisy directories (node_modules, __pycache__, .venv, …). Dotfiles and dot-directories are tracked unless excluded by the above rules. ``.cursorrules``, ``.editorconfig``, ``.eslintrc`` etc. are intentional project configuration; the blanket dot-skip that Git-adjacent tools inherited is not carried forward here. Paths use POSIX separators regardless of host OS for cross-platform reproducibility. Performance note: ``os.walk`` with in-place ``dirnames`` pruning is used instead of ``pathlib.rglob`` so that large noisy directories are never descended into. The stat cache further skips re-hashing files whose ``(mtime, size)`` is unchanged since the last walk. Ignore-pattern fast path: patterns are compiled into a single combined regex (see :func:`_build_filename_filter`) that is evaluated against the bare filename once per file. For the builtin secrets blocklist (9 simple ``*.ext`` / ``name`` patterns with no ``/``), this replaces 9 separate ``fnmatch`` calls with one ``re.search`` call — a ~10× speedup at 75 k scale that brings warm 1-file-change latency from ~850 ms to < 200 ms. Files whose filename can't possibly match any pattern skip ``is_ignored`` entirely; files that might match (rare) fall through to the full check. """ files, _ = walk_workdir_with_dirs(workdir) return files def compute_snapshot_id( manifest: Manifest, directories: list[str] | None = None, ) -> str: """Return sha256 of the sorted ``path NUL object_id`` pairs and directory paths. The null-byte separator prevents collisions from filenames or object IDs that contain the previous ``|`` / ``:`` separators. Sorting ensures two identical working trees always produce the same snapshot_id, regardless of filesystem traversal order. When *directories* is provided (non-empty), directory paths are appended to the hash payload so that a directory rename produces a different snapshot ID even when file contents are unchanged. Callers that do not yet track directories may omit the argument; the resulting ID is identical to the pre-directory-tracking behaviour. """ parts = sorted(f"{path}{_SEP}{oid}" for path, oid in manifest.items()) if directories: # Prefix directory entries with "dir" so they occupy a distinct namespace # from file entries and cannot collide with path/oid pairs. parts.extend(f"dir{_SEP}{d}" for d in sorted(directories)) payload = _SEP.join(parts).encode() return hashlib.sha256(payload).hexdigest() def detect_directory_renames( deleted_dirs: set[str], added_dirs: set[str], last_manifest: Manifest, current_manifest: Manifest, ) -> list[tuple[str, str]]: """Return ``[(old_dir, new_dir)]`` pairs detected from manifest diffs. A directory rename is inferred when all files that were under *old_dir* in *last_manifest* appear under *new_dir* in *current_manifest* with identical object IDs (same content, different path). Empty directories and directories whose file sets do not match any added directory are not returned. The heuristic is conservative: only 1-to-1 renames are reported. If multiple added directories share the same file set (unusual but possible), the match is ambiguous and no rename is emitted for that pair. """ renames: list[tuple[str, str]] = [] matched_new: set[str] = set() for old_dir in sorted(deleted_dirs): prefix = old_dir + "/" old_files = { path[len(prefix):]: oid for path, oid in last_manifest.items() if path.startswith(prefix) } if not old_files: continue # empty dir — can't match by content candidates = [ new_dir for new_dir in sorted(added_dirs) if new_dir not in matched_new ] for new_dir in candidates: new_prefix = new_dir + "/" new_files = { path[len(new_prefix):]: oid for path, oid in current_manifest.items() if path.startswith(new_prefix) } if new_files == old_files: renames.append((old_dir, new_dir)) matched_new.add(new_dir) break return renames def diff_workdir_vs_snapshot( workdir: pathlib.Path, last_manifest: Manifest, last_directories: list[str] | None = None, ) -> tuple[set[str], set[str], set[str], set[str], set[str], set[str]]: """Compare *workdir* against *last_manifest* from the previous commit. Returns a tuple of six disjoint path sets: - ``added`` — files in *workdir* absent from *last_manifest*. - ``modified`` — files present in both but with a differing sha256 hash. - ``deleted`` — files in *last_manifest* absent from *workdir*. - ``untracked`` — non-empty only when *last_manifest* is empty (first commit): every file in *workdir* is untracked. - ``added_dirs`` — directories present in *workdir* but not in *last_directories*. - ``deleted_dirs``— directories in *last_directories* absent from *workdir*. All paths use POSIX separators for cross-platform reproducibility. """ if not workdir.exists(): return ( set(), set(), set(last_manifest.keys()), set(), set(), set(last_directories or []), ) current_manifest, current_dirs = walk_workdir_with_dirs(workdir) current_paths = set(current_manifest.keys()) last_paths = set(last_manifest.keys()) if not last_paths: return set(), set(), set(), current_paths, set(current_dirs), set() added = current_paths - last_paths deleted = last_paths - current_paths common = current_paths & last_paths modified = {p for p in common if current_manifest[p] != last_manifest[p]} # A file that was tracked in the last snapshot but is now listed in # .museignore and still present on disk is not "deleted" — it has been # intentionally moved out of tracking. Reporting it as deleted would # block checkout, pollute status output, and cause stash pop to unlink it. # Only files that are genuinely absent from the working tree are deleted. if deleted: ignore_patterns = load_ignore_patterns(workdir) deleted = { p for p in deleted if not (is_ignored(p, ignore_patterns) and (workdir / p).exists()) } last_dirs_set = set(last_directories or []) current_dirs_set = set(current_dirs) added_dirs = current_dirs_set - last_dirs_set deleted_dirs = last_dirs_set - current_dirs_set return added, modified, deleted, set(), added_dirs, deleted_dirs def compute_commit_id( parent_ids: list[str], snapshot_id: str, message: str, committed_at_iso: str, ) -> str: """Return sha256 of the commit's canonical inputs. Uses null bytes as field separators to prevent separator-injection attacks from commit messages, author names, or branch names containing ``|`` characters. Given the same arguments on two machines the result is identical. ``parent_ids`` is sorted before hashing so insertion order does not affect determinism. """ parts = [ _SEP.join(sorted(parent_ids)), snapshot_id, message, committed_at_iso, ] payload = _SEP.join(parts).encode() return hashlib.sha256(payload).hexdigest()