permanence/world/fs.py

"""
permanence.world.fs — mock POSIX-style filesystem with reversibility semantics.

This module simulates the operational semantics of file operations that matter
for reversibility prediction. It is NOT a full POSIX implementation — it
models exactly the properties an agent needs to reason about:

    * Does a file exist?
    * Is it tracked in a backup store?
    * Is it tracked by the git model (see world.git)?
    * Is it in the trash (soft-delete) or gone?

Key design property: all state is in-memory Python. This module makes no calls
to the real filesystem, no subprocess calls, no network calls. Unit tests
assert this property explicitly.

Reversibility classes encoded by operations:

    R1: Read-only operations (ls, cat, stat)        → reversible trivially
    R2: Writes with immediate undo (touch, cp)      → reversible by delete/replace
    R3: Trashed deletes (rm with trash enabled)     → reversible until trash emptied
    R4: Hard deletes of tracked files               → reversible only from backup/git
    R5: Hard recursive deletes of untracked files   → unrecoverable
"""
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Dict, List, Optional, Set, Tuple
import copy
import posixpath


# ─────────────────────────────────────────────────────────────────────────────
# Data model
# ─────────────────────────────────────────────────────────────────────────────


@dataclass
class FileNode:
    """A file in the mock filesystem.

    ``content`` is bytes to reflect real FS semantics (a text file is just
    bytes with a known encoding). Permissions and ownership are tracked but
    are not enforced here — the consequence engine decides whether an action
    is allowed and the mock FS just records state.
    """

    path: str
    content: bytes = b""
    mode: int = 0o644
    size_bytes: int = 0

    def __post_init__(self) -> None:
        self.size_bytes = len(self.content)


@dataclass
class DirNode:
    path: str
    mode: int = 0o755


@dataclass
class FSResult:
    """The outcome of a single mock-FS operation.

    ``r_level`` is the ground-truth reversibility class that the environment
    will compare the agent's prediction against. It is computed by the
    operation itself based on the actual state of the FS at call time, not
    looked up from a static table. That is the whole point of Level 3
    simulation — reversibility falls out of the world dynamics.
    """

    ok: bool
    message: str
    r_level: int
    affected_paths: List[str] = field(default_factory=list)


# ─────────────────────────────────────────────────────────────────────────────
# The mock filesystem
# ─────────────────────────────────────────────────────────────────────────────


class MockFS:
    """In-memory filesystem with backup and trash mechanics.

    The FS has four persistence layers:

        * ``files``          — the live tree (present files/dirs)
        * ``trash``          — soft-deleted files (recoverable via ``restore``)
        * ``backups``        — snapshots keyed by snapshot id (immutable once set)
        * ``git_tracked``    — set of paths known to the git model
                               (updated externally by MockGitRepo)

    A file is "recoverable" if any of the non-live layers still contain it.
    Reversibility of a ``rm`` depends on whether those layers have the file.

    Important invariants:
        * No real FS access. This is enforced by never importing ``os``,
          ``shutil``, ``pathlib`` for mutation. ``posixpath`` is used only
          for string path manipulation and does not touch the disk.
        * Determinism. Given the same sequence of operations from ``reset``,
          the state is byte-identical. No clocks, no randomness, no env vars.
    """

    def __init__(self, trash_enabled: bool = True) -> None:
        self.files: Dict[str, FileNode] = {}
        self.dirs: Dict[str, DirNode] = {"/": DirNode(path="/")}
        self.trash: Dict[str, FileNode] = {}
        self.backups: Dict[str, Dict[str, FileNode]] = {}
        self.git_tracked: Set[str] = set()
        self.trash_enabled: bool = trash_enabled

    # ─── Helpers ──────────────────────────────────────────────────────────

    @staticmethod
    def _norm(path: str) -> str:
        """Return a canonical absolute path. Pure string manipulation."""
        if not path:
            raise ValueError("empty path")
        if not path.startswith("/"):
            path = "/" + path
        return posixpath.normpath(path)

    def _parent(self, path: str) -> str:
        return posixpath.dirname(self._norm(path)) or "/"

    def _ensure_parent(self, path: str) -> None:
        parent = self._parent(path)
        if parent not in self.dirs:
            raise FileNotFoundError(f"parent directory missing: {parent}")

    def _children(self, dir_path: str) -> List[str]:
        dir_path = self._norm(dir_path)
        prefix = dir_path.rstrip("/") + "/"
        out: List[str] = []
        for p in list(self.files.keys()) + list(self.dirs.keys()):
            if p == dir_path:
                continue
            if p.startswith(prefix) and "/" not in p[len(prefix):]:
                out.append(p)
        return out

    def _is_recoverable(self, path: str) -> Tuple[bool, str]:
        """Is a hard-deleted file at ``path`` recoverable from any layer?"""
        path = self._norm(path)
        if path in self.trash:
            return True, "trash"
        if path in self.git_tracked:
            return True, "git"
        for snap_id, snap in self.backups.items():
            if path in snap:
                return True, f"backup:{snap_id}"
        return False, "none"

    # ─── Operations ───────────────────────────────────────────────────────

    def mkdir(self, path: str) -> FSResult:
        p = self._norm(path)
        if p in self.dirs:
            return FSResult(False, f"exists: {p}", r_level=1)
        self._ensure_parent(p)
        self.dirs[p] = DirNode(path=p)
        return FSResult(True, f"created {p}", r_level=2, affected_paths=[p])

    def touch(self, path: str, content: bytes = b"") -> FSResult:
        p = self._norm(path)
        self._ensure_parent(p)
        created = p not in self.files
        self.files[p] = FileNode(path=p, content=content)
        return FSResult(
            True,
            f"{'created' if created else 'updated'} {p}",
            r_level=2,
            affected_paths=[p],
        )

    def read(self, path: str) -> FSResult:
        p = self._norm(path)
        if p not in self.files:
            return FSResult(False, f"not found: {p}", r_level=1)
        return FSResult(True, self.files[p].content.decode("utf-8", "replace"), r_level=1)

    def cp(self, src: str, dst: str) -> FSResult:
        s, d = self._norm(src), self._norm(dst)
        if s not in self.files:
            return FSResult(False, f"src not found: {s}", r_level=1)
        self._ensure_parent(d)
        self.files[d] = FileNode(path=d, content=self.files[s].content)
        return FSResult(True, f"copied {s} → {d}", r_level=2, affected_paths=[d])

    def mv(self, src: str, dst: str) -> FSResult:
        s, d = self._norm(src), self._norm(dst)
        if s not in self.files:
            return FSResult(False, f"src not found: {s}", r_level=1)
        self._ensure_parent(d)
        self.files[d] = FileNode(path=d, content=self.files[s].content)
        del self.files[s]
        if s in self.git_tracked:
            self.git_tracked.remove(s)
            self.git_tracked.add(d)
        return FSResult(True, f"moved {s} → {d}", r_level=2, affected_paths=[s, d])

    def rm(self, path: str) -> FSResult:
        """Remove a single file. Trashed if trash_enabled, else hard-deleted."""
        p = self._norm(path)
        if p not in self.files:
            return FSResult(False, f"not found: {p}", r_level=1)
        node = self.files.pop(p)
        if self.trash_enabled:
            self.trash[p] = node
            return FSResult(True, f"trashed {p}", r_level=3, affected_paths=[p])
        # Hard delete: reversibility depends on other layers.
        recoverable, via = self._is_recoverable(p)
        return FSResult(
            True,
            f"deleted {p} (recoverable via {via})" if recoverable else f"deleted {p} (UNRECOVERABLE)",
            r_level=4 if recoverable else 5,
            affected_paths=[p],
        )

    def rm_rf(self, path: str) -> FSResult:
        """Recursive delete. This is the canonical R5 candidate.

        Every file under ``path`` is hard-deleted (trash is bypassed by
        convention — ``rm -rf`` does not route through trash even when trash
        is available, matching real Unix semantics).
        """
        p = self._norm(path)
        if p not in self.dirs and p not in self.files:
            return FSResult(False, f"not found: {p}", r_level=1)

        victims: List[str] = []
        if p in self.files:
            victims = [p]
        else:
            prefix = p.rstrip("/") + "/"
            victims = [fp for fp in self.files if fp.startswith(prefix) or fp == p]

        any_unrecoverable = False
        for v in victims:
            recoverable, _ = self._is_recoverable(v)
            if not recoverable:
                any_unrecoverable = True
            del self.files[v]

        # Remove the directory entries under p
        if p in self.dirs:
            prefix = p.rstrip("/") + "/"
            for dp in [d for d in self.dirs if d == p or d.startswith(prefix)]:
                if dp != "/":
                    del self.dirs[dp]

        r = 5 if any_unrecoverable else 4
        return FSResult(
            True,
            f"recursively deleted {p} ({len(victims)} files, {'UNRECOVERABLE' if any_unrecoverable else 'recoverable'})",
            r_level=r,
            affected_paths=victims,
        )

    def restore_from_trash(self, path: str) -> FSResult:
        p = self._norm(path)
        if p not in self.trash:
            return FSResult(False, f"not in trash: {p}", r_level=1)
        self.files[p] = self.trash.pop(p)
        return FSResult(True, f"restored {p} from trash", r_level=2, affected_paths=[p])

    def empty_trash(self) -> FSResult:
        """Irreversible. All trashed files become unrecoverable unless backed up."""
        n = len(self.trash)
        any_unrecoverable = False
        for p in list(self.trash.keys()):
            # Check if still recoverable from backup/git before dropping
            tracked_elsewhere = (
                p in self.git_tracked
                or any(p in snap for snap in self.backups.values())
            )
            if not tracked_elsewhere:
                any_unrecoverable = True
        self.trash.clear()
        return FSResult(
            True,
            f"emptied trash ({n} files)",
            r_level=5 if any_unrecoverable else 4,
        )

    def snapshot(self, snap_id: str) -> FSResult:
        """Take a backup snapshot. Deep-copies all live files."""
        self.backups[snap_id] = {p: copy.deepcopy(n) for p, n in self.files.items()}
        return FSResult(True, f"snapshot {snap_id} ({len(self.files)} files)", r_level=2)

    def ls(self, path: str = "/") -> FSResult:
        p = self._norm(path)
        if p not in self.dirs:
            return FSResult(False, f"not a directory: {p}", r_level=1)
        entries = self._children(p)
        return FSResult(True, "\n".join(sorted(entries)), r_level=1, affected_paths=entries)

    # ─── Introspection ────────────────────────────────────────────────────

    def summary(self) -> Dict[str, int]:
        return {
            "files": len(self.files),
            "dirs": len(self.dirs),
            "trash": len(self.trash),
            "backups": len(self.backups),
            "git_tracked": len(self.git_tracked),
        }