Spaces:
Paused
Paused
File size: 3,218 Bytes
5e0b58b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | """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)
|