src/storage/layout.py

"""Canonical filesystem layout for operator, experiment, and dev artifacts."""

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path


def validate_scope_transition(source: str, target: str, operation: str) -> None:
    """Reject storage transitions that could merge uncontrolled state."""
    if source.upper() == "DEV" and target.upper() == "OPERATOR":
        raise ValueError("DEV to OPERATOR transitions are forbidden")
    if (
        source.upper() == "EXPERIMENT"
        and target.upper() == "OPERATOR"
        and operation.lower() == "merge"
    ):
        raise ValueError("EXPERIMENT to OPERATOR merge is forbidden")
    if (
        source.upper() == "OPERATOR"
        and target.upper() == "EXPERIMENT"
        and operation.lower() not in {"snapshot", "fork"}
    ):
        raise ValueError("OPERATOR to EXPERIMENT requires snapshot or fork")
    if source.upper() not in {
        "DEV",
        "EXPERIMENT",
        "OPERATOR",
    } or target.upper() not in {"DEV", "EXPERIMENT", "OPERATOR"}:
        raise ValueError(f"Unknown storage scope transition: {source} -> {target}")


@dataclass(frozen=True, slots=True)
class StorageLayout:
    """Resolve controlled storage roots without crossing the workspace root."""

    root: Path

    def __post_init__(self) -> None:
        object.__setattr__(self, "root", self.root.resolve())

    @property
    def operator_state(self) -> Path:
        return self.root / "operator" / "state.b5d"

    @property
    def operator_journal(self) -> Path:
        return self.root / "operator" / "journal"

    @property
    def operator_action_journal(self) -> Path:
        """Durable hash-linked action audit path within operator storage."""
        return self.operator_journal / "actions.jsonl"

    @property
    def operator_checkpoints(self) -> Path:
        return self.root / "operator" / "checkpoints"

    @property
    def dev_disposable(self) -> Path:
        return self.root / "dev" / "disposable"

    def experiment(self, experiment_id: str) -> Path:
        if (
            not experiment_id.startswith("EXP-")
            or "/" in experiment_id
            or "\\" in experiment_id
        ):
            raise ValueError("Experiment ID must be a safe EXP-* name")
        candidate = self.root / "experiment" / experiment_id
        try:
            candidate.resolve().relative_to(self.root)
        except ValueError as exc:
            raise ValueError("Experiment path escapes storage root") from exc
        return candidate

    def experiment_state(self, experiment_id: str) -> Path:
        return self.experiment(experiment_id) / "state"

    def experiment_data(self, experiment_id: str) -> Path:
        return self.experiment(experiment_id) / "DATA"

    def experiment_evidence(self, experiment_id: str) -> Path:
        return self.experiment(experiment_id) / "EVID"

    def ensure_directories(self) -> None:
        """Create only controlled roots, never experiment data."""
        for directory in (
            self.operator_journal,
            self.operator_checkpoints,
            self.dev_disposable,
        ):
            directory.mkdir(parents=True, exist_ok=True)