orchestrata_ultimate.openspec.repo

Reference › orchestrata_ultimate › openspec › repo

repo gives sandboxed read and write access to an OpenSpec tree (SPEC-104 FR-004, FR-005, FR-007 and FR-008). OpenSpecRepo is built on a root directory and resolves every path through _safe, which confines writes to the detected spec root (SEC-104-001). capabilities lists directories that contain a spec.md. read_capability parses one with parse and returns an empty document when the file is absent. write_capability renders a document with render and writes it, creating directories as needed. add_requirement appends a requirement to a capability and rewrites its file. create_change scaffolds a changes/<id>/ folder with a proposal, a task list and optional delta specs, and refuses to overwrite an existing change (SEC-104-002, ERR-104-005). OpenSpecError is raised for sandbox violations and change-id collisions. Both classes carry @traces for REQ-104-010. The package imports it.

At a glance

Class diagram

Dependency diagram

API

Classes

OpenSpecError

Bases: Exception · Decorators: traces(spec='104', story='US-104-001', reqs=['REQ-104-010']) · Line 24

Raised for sandbox violations and change-id collisions.

OpenSpecRepo

Decorators: traces(spec='104', story='US-104-001', reqs=['REQ-104-010']) · Line 29

Sandboxed read/write access to an OpenSpec tree rooted at root.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/openspec/repo.py

"""OpenSpec repository operations (SPEC-104 FR-004/005/007/008).

Read/create/edit capability specs and scaffold change proposals, with all writes
confined to the detected spec root (SEC-104-001) and a refusal to overwrite an
existing change (SEC-104-002 / ERR-104-005).
"""

from __future__ import annotations

from pathlib import Path

from orchestrata_ultimate.core.traceability import traces
from orchestrata_ultimate.openspec.models import Delta, OpenSpecDoc, Requirement
from orchestrata_ultimate.openspec.parse import parse_spec
from orchestrata_ultimate.openspec.render import (
    render_delta,
    render_proposal,
    render_spec,
    render_tasks,
)


@traces(spec="104", story="US-104-001", reqs=["REQ-104-010"])
class OpenSpecError(Exception):
    """Raised for sandbox violations and change-id collisions."""


@traces(spec="104", story="US-104-001", reqs=["REQ-104-010"])
class OpenSpecRepo:
    """Sandboxed read/write access to an OpenSpec tree rooted at ``root``."""

    def __init__(self, root: str | Path) -> None:
        self.root = Path(root).resolve()
        # specs live under <root>/specs when present, else directly under root.
        self.specs_dir = (
            self.root / "specs" if (self.root / "specs").is_dir() else self.root
        )
        self.changes_dir = self.root / "changes"

    @traces(spec="104", story="US-104-001", reqs=["REQ-104-010"])
    def _safe(self, path: Path) -> Path:
        resolved = (
            (self.root / path).resolve() if not path.is_absolute() else path.resolve()
        )
        if resolved != self.root and self.root not in resolved.parents:
            raise OpenSpecError(f"path escapes spec root: {path}")
        return resolved

    @traces(spec="104", story="US-104-001", reqs=["REQ-104-001"])
    def capabilities(self) -> list[str]:
        """List capability names (directories with a spec.md)."""
        if not self.specs_dir.is_dir():
            return []
        return sorted(
            d.name
            for d in self.specs_dir.iterdir()
            if d.is_dir() and (d / "spec.md").is_file()
        )

    @traces(spec="104", story="US-104-001", reqs=["REQ-104-004"])
    def read_capability(self, name: str) -> OpenSpecDoc:
        """Parse the capability's spec.md (empty doc if absent)."""
        spec = self._safe(self.specs_dir / name / "spec.md")
        text = spec.read_text(encoding="utf-8") if spec.is_file() else ""
        return parse_spec(text, capability=name)

    @traces(spec="104", story="US-104-001", reqs=["REQ-104-006"])
    def write_capability(self, doc: OpenSpecDoc) -> Path:
        """Render and write a capability spec.md (creating dirs as needed)."""
        spec = self._safe(self.specs_dir / doc.capability / "spec.md")
        spec.parent.mkdir(parents=True, exist_ok=True)
        spec.write_text(render_spec(doc), encoding="utf-8")
        return spec

    @traces(spec="104", story="US-104-001", reqs=["REQ-104-008"])
    def add_requirement(self, capability: str, requirement: Requirement) -> Path:
        """Append a requirement to a capability and rewrite its spec.md."""
        doc = self.read_capability(capability)
        doc.capability = capability
        doc.requirements.append(requirement)
        return self.write_capability(doc)

    @traces(spec="104", story="US-104-001", reqs=["REQ-104-007"])
    def create_change(
        self,
        change_id: str,
        why: str,
        what: list[str],
        tasks: list[str],
        deltas: list[Delta] | None = None,
        capability: str = "",
    ) -> Path:
        """Scaffold changes/<id>/ with proposal.md, tasks.md and an optional delta.

        Refuses to overwrite an existing change directory (ERR-104-005).
        """
        change_dir = self._safe(self.changes_dir / change_id)
        if change_dir.exists():
            raise OpenSpecError(f"change already exists: {change_id}")
        change_dir.mkdir(parents=True)
        (change_dir / "proposal.md").write_text(
            render_proposal(why, what), encoding="utf-8"
        )
        (change_dir / "tasks.md").write_text(render_tasks(tasks), encoding="utf-8")
        if deltas and capability:
            delta_spec = self._safe(change_dir / "specs" / capability / "spec.md")
            delta_spec.parent.mkdir(parents=True, exist_ok=True)
            delta_spec.write_text(render_delta(deltas), encoding="utf-8")
        return change_dir