Persist registry to disk. Raises ``StepValidationError`` with a clear message on filesystem errors (read-only fs, permission denied, ...) so callers can surface a clean error to the user rather than an unhandled ``OSError``.
(self)
| 675 | return default_registry |
| 676 | |
| 677 | def save(self) -> None: |
| 678 | """Persist registry to disk. |
| 679 | |
| 680 | Raises ``StepValidationError`` with a clear message on filesystem |
| 681 | errors (read-only fs, permission denied, ...) so callers can surface |
| 682 | a clean error to the user rather than an unhandled ``OSError``. |
| 683 | """ |
| 684 | if self._has_symlinked_parent() or self.registry_path.is_symlink(): |
| 685 | raise StepValidationError( |
| 686 | "Refusing to write step registry through a symlinked path." |
| 687 | ) |
| 688 | try: |
| 689 | self.steps_dir.mkdir(parents=True, exist_ok=True) |
| 690 | with open(self.registry_path, "w", encoding="utf-8") as f: |
| 691 | json.dump(self.data, f, indent=2) |
| 692 | except OSError as exc: |
| 693 | raise StepValidationError( |
| 694 | f"Failed to write step registry at {self.registry_path}: {exc}" |
| 695 | ) from exc |
| 696 | |
| 697 | def add(self, step_id: str, metadata: dict[str, Any]) -> None: |
| 698 | """Add or update an installed step entry.""" |