Function _restore_hardlinked_dir

openkb/mutation.py:260–305

O(touched) restore for a hardlinked directory backup. The backup was built with ``os.link``, so live files the mutation never touched still share the backup's inode — leave them. Only files the mutation changed need work: new files (no backup counterpart) are removed, modified files

(backup: Path, target: Path)

def _restore_hardlinked_dir(backup: Path, target: Path) -> None:
    """O(touched) restore for a hardlinked directory backup.

    The backup was built with ``os.link``, so live files the mutation never
    touched still share the backup's inode — leave them. Only files the
    mutation changed need work: new files (no backup counterpart) are removed,
    modified files (atomic temp+replace → new inode) and deleted files are
    restored from the backup's pre-mutation bytes. This avoids recopying the
    whole tree on rollback — the cost that bit ``.openkb/files`` (the blob
    store) and large concept/entity trees on every failed add.

    Degrades gracefully to a full copy if the backup isn't actually hardlinked
    (e.g. the EXDEV/EACCES fallback fired at snapshot time): every file then has
    a different inode, so every file is treated as modified and recopied.
    """

    def _file_key(path: Path) -> tuple[int, int]:
        st = path.stat()  # follows symlinks; these trees hold regular files only
        return (st.st_dev, st.st_ino)

    backup_files = {p.relative_to(backup): p for p in backup.rglob("*") if p.is_file()}

    # Pass 1: remove new + modified live regular files; leave untouched ones
    # (they share the backup inode) in place.
    if target.exists():
        for live in list(target.rglob("*")):
            if not live.is_file():
                continue
            counterpart = backup_files.get(live.relative_to(target))
            if counterpart is None or _file_key(live) != _file_key(counterpart):
                live.unlink()

    # Pass 2: restore modified + deleted files from backup.
    for rel, src in backup_files.items():
        dest = target / rel
        if not dest.exists() or _file_key(dest) != _file_key(src):
            dest.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(src, dest)

    # Pass 3: prune directories the mutation created that are now empty.
    if target.exists():
        for d in sorted(
            (p for p in target.rglob("*") if p.is_dir()), key=lambda p: len(p.parts), reverse=True
        ):
            if not (backup / d.relative_to(target)).exists() and not any(d.iterdir()):
                d.rmdir()


def snapshot_paths(