root

class Scratch:
    """One process's handle on a scratch root; see the module for the layout."""

    def __init__(
        self,
        root: Path,
        *,
        max_size_gb: float | None = None,
        free_space_ratio: float | None = None,
    ) -> None:
        """Open or create the scratch at ``root`` and reclaim what dead processes left.

        Args:
            root: The scratch directory.
            max_size_gb: A hard cap on the cache, persisted into ``settings.toml``.
            free_space_ratio: The fraction of releasable disk the cache may occupy,
                persisted into ``settings.toml``.

        An index of an older layout is migrated in place where the change
        allows; otherwise the scratch is emptied and rebuilt.
        """
        self.root = root.expanduser().resolve()
        self.root.mkdir(parents=True, exist_ok=True)
        (self.root / ".gitignore").write_text("*\n")
        for subdirectory in (
            self.objects_dir,
            self.state_locks_dir,
            self.call_locks_dir,
            self.resource_locks_dir,
            self.root / "live",
        ):
            subdirectory.mkdir(parents=True, exist_ok=True)
        self.settings = Settings.load_or_create(
            self.root,
            max_size_gb=max_size_gb,
            free_space_ratio=free_space_ratio,
        )
        self.database = Database(self.root / INDEX_FILENAME)
        if self.database.rebuilt:
            remove_tree(self.objects_dir)
            self.objects_dir.mkdir()
        self.file_hashes = FileHashes(connect(self.root / INDEX_FILENAME))
        self.liveness = Liveness.acquire(self.locks_dir)
        self.leaked_pins: deque[ContentHash] = deque()
        self.closed = False
        self._finalizer = weakref.finalize(
            self,
            _close,
            self.database,
            self.file_hashes,
            self.liveness,
            self.live_dir,
        )
        self.reconcile()

    # --- layout -------------------------------------------------------------

    @property
    def objects_dir(self) -> Path:
        """Where every sealed object lives."""
        return self.root / "objects"

    @property
    def locks_dir(self) -> Path:
        """Where every process's liveness lock lives."""
        return self.root / "locks"

    @property
    def state_locks_dir(self) -> Path:
        """Where every state key's lock lives."""
        return self.locks_dir / "state"

    @property
    def call_locks_dir(self) -> Path:
        """Where the lock of every call being computed lives."""
        return self.locks_dir / "call"

    @property
    def resource_locks_dir(self) -> Path:
        """Where the lock of every named resource lives."""
        return self.locks_dir / "resource"

    @property
    def live_dir(self) -> Path:
        """Where this process's workspaces live."""
        return self.root / "live" / self.liveness.name

    def object_dir(self, hash: ContentHash) -> Path:
        """The directory the object ``hash`` occupies."""
        return self.objects_dir / hash.hex

    # --- trees --------------------------------------------------------------

    def hash_tree(self, path: Path | str) -> ContentHash:
        """The merkle hash of the tree at ``path``, remembering file hashes by stat.

        Raises:
            FileNotFoundError: If ``path`` does not exist.
        """
        return hash_tree(path, self.file_hashes)

    def workspace(self) -> Workspace:
        """A fresh directory to build a tree in; gone when released or collected."""
        return Workspace(self.live_dir)

    # --- objects, names, pins, locks ----------------------------------------

    def seal(
        self,
        key: CallKey,
        workspace: Workspace,
        stems: list[str],
        compute_seconds: float,
    ) -> list[Pinned]:
        """Seal ``workspace/<stem>`` per stem as the results of the call ``key``, pinned."""
        return seal.seal_call(self, key, workspace, stems, compute_seconds)

    def memo(self, key: CallKey, count: int = 1) -> Memo:
        """What this scratch has for the call ``key``, whose call returns ``count`` results."""
        return call_memo(self, key, count)

    def hold(self, name: str) -> Hold:
        """A hold on the object ``name`` points at, and on whatever it is pointed at next."""
        return Hold(self, name)

    def lock_state(self, key: str, timeout_seconds: float = 30.0) -> StateLock:
        """Exclusive access to the state ``key``.

        Raises:
            RecipeError: If the lock is still held elsewhere after ``timeout_seconds``.
        """
        return StateLock(self, key, timeout_seconds)

    def lock_call(self, key: CallKey) -> KeyLock:
        """The right to compute the call ``key``, waiting for whoever holds it to finish."""
        return KeyLock(self.call_locks_dir, key.hex).acquire(None)

    def lock_resource(self, name: str) -> KeyLock:
        """A handle on the machine-wide resource ``name``, not yet acquired."""
        return KeyLock(self.resource_locks_dir, name)

    # --- in-flight steps ----------------------------------------------------

    def steps(self) -> list[StepRow]:
        """Every step a living process has open in this scratch, oldest first."""
        with self.database.transaction() as db:
            rows = steps.all_rows(db)
        return [row for row in rows if self._owner_alive(row)]

    def _owner_alive(self, row: StepRow) -> bool:
        return owner_alive(self.locks_dir, row.owner_pid, row.owner_boot)

    def evict(self) -> None:
        """Delete the least valuable unheld objects until the cache fits its budget."""
        eviction.evict(self)

    # --- lifecycle ----------------------------------------------------------

    def reconcile(self) -> None:
        """Make the index agree with the disk and forget what dead processes held."""
        for entry in (self.root / "live").iterdir():
            if entry.is_dir() and lock_is_free(self.locks_dir / entry.name):
                remove_tree(entry)
        present = {
            ContentHash.from_hex(entry.name)
            for entry in self.objects_dir.iterdir()
            if entry.is_dir() and len(entry.name) == 64
        }
        with self.database.transaction() as db:
            indexed = {row.hash for row in objects.all_rows(db)}
            for hash in present - indexed:
                objects.seed(db, hash, tree_size(self.object_dir(hash)))
            for hash in indexed - present:
                objects.delete(db, hash)
            for name, hash in names.all_rows(db).items():
                if hash not in present:
                    names.delete(db, name)
            for pin in pins.all_rows(db):
                if pin.hash not in present or not owner_alive(
                    self.locks_dir, pin.owner_pid, pin.owner_boot
                ):
                    pins.delete(db, pin.hash, pin.owner_boot)
            for row in steps.all_rows(db):
                if not self._owner_alive(row):
                    steps.delete(db, row.id)
        self._reap_free_locks()

    def _reap_free_locks(self) -> None:
        """Delete the liveness and call lock files nobody holds.

        A state or resource lock file is left alone: unlinking one races a
        holder that has opened it but not yet flocked it, and two writers of one
        state lose an update (two holders of the GPU run out of memory), where
        two claimants of one call only duplicate it.
        """
        for directory in (self.locks_dir, self.call_locks_dir):
            for entry in directory.iterdir():
                if entry.is_file() and entry != self.liveness.path and lock_is_free(entry):
                    entry.unlink(missing_ok=True)

    def close(self) -> None:
        """Release this process's hold: its workspaces, its liveness, its index connections."""
        self.closed = True
        self._finalizer()

    def __repr__(self) -> str:
        """``Scratch(<root>)``."""
        return f"Scratch({str(self.root)!r})"